pg-promise :取消pg- query -stream发起的查询

Question

我有一个postgresql表，每天都有成千上万的时间序列数据。我有一个应用程序，允许用户检索这些数据。根据时间范围的不同，查询可能需要200毫秒到30秒的时间，因此这些查询必须是可取消的，以避免生产环境中的无用负载。

由于有数十亿的数据，使用流来检索它们是不可避免的。

因此，我设法获得了一个像pg-promise文档中的图形那样的数据流工作端点，并通过关闭pg-query-stream中的光标使其可取消。

下面是在此端点中执行的操作示例(在构建查询后调用dataStream())：

const pgp = require("pg-promise")();
const QueryStream = require("pg-query-stream");

const db = pgp({
  host: "1.2.3.4",
  port: 5432,
  database: "db",
  user: "user",
  password: "password",
  max: 2,
});

// query is an SQL string
dataStream(query, req, res, next) {
  const qs = new QueryStream(query);

  // "close" event is triggered on client request cancelation
  req.on("close", () => {
    qs.destroy();
  });

  return db.stream(qs, s => {
    s.pipe(JSONStream.stringify()).pipe(res);
    s.on("error", error => handleError(error));
  })
  .catch(error => handleError(error, query));
}

它对一些调用很有效，但在某些时候(快速执行8到10个调用以检查可取消性)，应用程序会使用以下堆栈崩溃：

\node_modules\pg-promise\node_modules\pg\lib\client.js:346
    if (self.activeQuery.name) {
                         ^

TypeError: Cannot read property 'name' of null
    at Connection.<anonymous> (\node_modules\pg-promise\node_modules\pg\lib\client.js:346:26)
    at Connection.emit (events.js:311:20)
    at Socket.<anonymous> (\node_modules\pg-promise\node_modules\pg\lib\connection.js:120:12)
    at Socket.emit (events.js:311:20)
    at addChunk (_stream_readable.js:294:12)
    at readableAddChunk (_stream_readable.js:275:11)
    at Socket.Readable.push (_stream_readable.js:209:10)
    at TCP.onStreamRead (internal/stream_base_commons.js:186:23)

因此，我怀疑调用qs.destroy()来关闭流不是正确的方法，即使游标在服务器端被很好地销毁了。

感谢node-postgres和pg-promise开发人员为您所做的工作。

Answer 1

对于那些感兴趣的人，经过多次尝试，我找到了一个可行的解决方案。它还解决了我遇到的另一个问题:通过发送垃圾邮件检查请求的可取消性，我注意到池中的一些客户端永远挂起，再也不会返回，导致池已满，新请求永远挂起。

我认为这可以通过以下事实来解释：res被管道传输到流中，并且当请求被取消时，可读流永远不会被消耗和挂起。我代码中的另一个问题是req.on("close",并不总是被触发。为了解决这个问题，我找到了一个名为on-finished的模块，它的作用就像想要的一样。

此外，调用qs.destroy()也不是正确的方法。经过长时间的调试，没有未处理错误的最一致的方法是从pgp的Database.connect()获取连接对象，并通过调用connection.done()结束查询。

所以这是我的解决方案：

const pgp = require("pg-promise")();
const QueryStream = require("pg-query-stream");
const JSONStream = require("JSONStream");
const onFinished = require("on-finished");

const db = pgp({
  host: "1.2.3.4",
  port: 5432,
  database: "db",
  user: "user",
  password: "password",
  max: 2,
});

// query is an SQL string
async function dataStream(query, req, res, next) {
    try {
      if (query instanceof Object) {
        query = query.toString();
      }
      const connection = await db.connect();
      const qs = new QueryStream(query, [], {highWaterMark: 4000});
      const streamData = connection.client.query(qs);

      onFinished(res, () => {
        // Calling .done() to end the connection on request close.
        // Weirdly I sometimes get an error if I do not provide a callback.
        connection.done(error => {
          log.error(error);
        });
      });
      streamData.pipe(JSONStream.stringify()).pipe(res);

      streamData.on("error", error => {
        next(error);
      });
    } catch (error) {
      next(error);
    }
  }