NodeJs Streams、pipelines和https post

Question

我需要一点理智的检查。

我正在运行节点11.10.1

我有一个使用nodejs oracledb库从oracle db读取数据的进程。有一个流函数，我执行select *并将结果以10k个对象为一批进行流式处理。然后，我通过https将此数据发布到索引器。对象流被注入到管道函数中。

我使用下面的代码已经有一段时间了。我正在尝试调试吞吐量。有时，我可以看到通过此管道每秒处理大约2k个文档。大多数时候，我看到的是<150。在我开始调试我的索引服务器之前。我想确保这些函数编码正确。

  async function streamReindex(databaseStream) {
    let pipeline = util.promisify(stream.pipeline)
    await pipeline(
      selectStream,//  "oracledb": "^4.0.0", stream function
      camelize.camelizeStream(), //"camelize2": "^1.0.0", library wrapped in ,"through2": "^3.0.1" library to make it an object stream
      JSONStream.stringify(), //"JSONStream": "^1.3.5"
      reindexClient.streamReindex(core)
    )
  }

// reindexClient code.

  function streamReindex(core) {
    const updateUrl = baseUrl + core + '/update'
    const options = {
      method: 'POST',
      headers: {
        'Content-Type': 'application/json'
      },
      'auth': `${user.username}:${user.password}`,
    }
    let postStream = https.request(updateUrl, options, (res) => {
      let response = {
        status: {
          code: res.statusCode,
          message: res.statusMessage
        },
        headers: res.headers,
      }
      if (res.statusCode !== 200) {
        postStream.destroy(new Error(JSON.stringify(response)))
      }
    })
    postStream.on('error', (err)=>{
      throw new Error(err)
    })
    postStream.on('socket', (socket) => {
      socket.setKeepAlive(true, 240000)
    })
    return postStream
  }


  async function selectStream(sql, bindings = [], fetchSize = 
     fetchArraySize) {
     let connection = await knex.client.acquireConnection()

    log.info(`Fetch size is set to ${fetchSize}`)
    let select = connection.queryStream(sql, bindings, {
      fetchArraySize: fetchSize,
      outFormat: outFormat
    })

    select.on('error', (err) => {
      log.error('Oracle Error Event', err)
      knex.client.releaseConnection(connection)
    })

    select.on('close', () => {
      log.info('Oracle Close Event')
      knex.client.releaseConnection(connection)
      select = null
      connection = null
    })

    return select
  }

如果我从管道中删除reindexClient.streamReindex(核心)函数。我看到每秒大约5k个对象的吞吐量。我正在研究streams的高水位线功能，但似乎不知道如何在postStream上应用它。如果我console.log post流，它也不会说它处于对象模式。这意味着它的高水位线是以字节为单位的，我相信它的阈值很低。

如果你需要更多的信息，我会尽可能多的提供。

Answer 1

因此，我通过自己实现JSONStream.stringify()函数，每秒增加了近500个/docs的吞吐量。因为它不允许我设置水印。一旦我这样做了，我就能够真正地提高它，但我可以使用我所有的内存。使用下面的代码进行设置，我得到了稳定的内存占用，并且具有更好的吞吐量。我还去掉了through2库，并在stringify转换中加入了camelize功能。大部分代码和解释都可以在这里找到：https://blog.dmcquay.com/2017/09/06/node-stream-db-results-with-transform.html

代码如下：

  function camelizeAndStringify() {
    let first = true
    const serialize = new Transform({
      objectMode: true,
      highWaterMark: 1000,
      transform(chunk, encoding, callback) {
        if (first) {
          this.push('[' + JSON.stringify(camelize(chunk)))
          first = false
        } else {
          this.push(',' + JSON.stringify(camelize(chunk)))
        }
        callback()
        chunk = null
      },
      flush(callback) {
        this.push(']')
        callback()
      }
    })
    return serialize
  }

Answer 2

虽然您的问题似乎不是oracledb的问题，但我将它放在这里，以便我可以格式化代码。您可能会从调整oracledb流中获得一些性能优势，例如：

   diff --git a/lib/queryStream.js b/lib/queryStream.js
   index 08ddc720..11953e4b 100644
   --- a/lib/queryStream.js
   +++ b/lib/queryStream.js
   @@ -24,7 +24,7 @@ const { Readable } = require('stream');
    class QueryStream extends Readable {

      constructor(rs) {
   -    super({ objectMode: true });
   +    super({ objectMode: true, highWaterMark: 64 });  // choose your own value
    this._fetching = false;
    this._numRows = 0;

允许将高水位线设置为queryStream()选项的PR将是受欢迎的。