NodeJs流、管道和https帖子

Question

我需要一点健全性检查。

我正在运行Node 11.10.1

我有一个使用nodejs oracledb库从oracledb读取的进程。有一个流式传输功能，我选择*并以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(核心(函数。我看到每秒大约5000个对象的吞吐量。我在研究流的高水印功能，但似乎不知道如何在postStream上应用它。如果我console.log发布流，它也不会说它处于对象模式。这意味着它的高水印是以字节为单位的，我相信它的阈值很低。

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

Answer 1

虽然您的问题似乎与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将是受欢迎的。

Answer 2

因此，我通过自己实现JSONStream.stringify((函数，使吞吐量每秒增加了近500/docs。因为它不允许我设置水印。一旦我做到了，我就能够真正提高它，但我可以利用我所有的记忆。使用以下代码进行设置使我获得了稳定的内存占用时间和更好的吞吐量。我还去掉了through2库，并将camalize功能放在我的字符串Transfrom中。大部分代码和说明可在此处找到：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
}