如何读取一行与node.js或javascript的文件延迟，而不是在非阻塞行为

How to read lines of a file with node.js or javascript with delay, not in non-blocking behavior?

本文关键字：延迟 文件 读取 何读取 一行 javascript js node      更新时间：2023-09-26

我正在读取node.js中的文件(300,000行)。我想将5000行的批量发送到另一个应用程序(Elasticsearch)来存储它们。因此，每当我读取完5000行，我想通过API将它们批量发送到Elasticsearch以存储它们，然后继续读取文件的其余部分并每5000行批量发送一次。

如果我想使用java(或任何其他阻塞语言，如C, c++， python等)来完成此任务，我会这样做:

int countLines = 0;
String bulkString = "";
BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream("filePath.txt")));
while ((currentLine = br.readLine()) != null) {
     countLines++;
     bulkString += currentLine;
     if(countLines >= 5000){
          //send bulkString to Elasticsearch via APIs
          countLines = 0;
          bulkString = "";
     }
}

如果我想对node.js做同样的事情，我会做:

var countLines = 0;
var bulkString = "";
var instream = fs.createReadStream('filePath.txt');
var rl = readline.createInterface(instream, outstream);
rl.on('line', function(line) {
     if(countLines >= 5000){
          //send bulkString to via APIs
          client.bulk({
          index: 'indexName',
          type: 'type',
          body: [bulkString]
          }, function (error, response) {
          //task is done
          });
          countLines = 0;
          bulkString = "";
     }
}

node.js的问题是它是非阻塞的，所以它在发送下一批行之前不会等待第一个API响应。我知道这对done.js来说是一个好处，因为它不需要等待I/O，但问题是它向Elasticsearch发送了太多的数据。因此，Elasticsearch的队列将被填满，它将抛出异常。

我的问题是我如何使node.js在继续读取下一行或将下一批行发送到Elasticsearch之前等待API的响应。

我知道我可以在Elasticsearch中设置一些参数来增加队列大小，但我对node.js的阻塞行为感兴趣。我很熟悉回调的概念，但是我想不出在这种情况下使用回调的方法来防止node.js以非阻塞模式调用Elasticsearch API。

皮埃尔的回答是正确的。我只是想提交一个代码，展示我们如何从node.js的非阻塞概念中受益，但同时，不要一次用太多的请求压倒Elasticsearch。

下面是一个伪代码，您可以使用它通过设置队列大小限制来为代码提供灵活性:

var countLines = 0;
var bulkString = "";
var queueSize = 3;//maximum of 3 requests will be sent to the Elasticsearch server
var batchesAlreadyInQueue = 0;
var instream = fs.createReadStream('filePath.txt');
var rl = readline.createInterface(instream, outstream);
rl.on('line', function(line) {
     if(countLines >= 5000){
          //send bulkString to via APIs
          client.bulk({
          index: 'indexName',
          type: 'type',
          body: [bulkString]
          }, function (error, response) {
               //task is done
               batchesAlreadyInQueue--;//we will decrease a number of requests that are already sent to the Elasticsearch when we hear back from one of the requests
               rl.resume();
          });
          if(batchesAlreadyInQueue >= queueSize){
               rl.pause();
          }
          countLines = 0;
          bulkString = "";
     }
}

在if之后使用rl.pause()，在//task is done之后使用rl.resume()。

请注意，在调用pause之后可能会有更多的行事件。

相关文章：