javascript – 异步队列,filestream结束如何知道何时完成

当使用async.queue和filestream时,我有一个小问题

>我有一个场景,我的filestream将完成
>我将fileRead设置为true
>然而队列将为空,已经被称为排水
>然后引导我的“完成”永远不会被调用

在我的filestream“结束”并且队列为空之后,说出“结束队列”的正确方法是什么？

var fs = require('fs'),util = require('util'),stream = require('stream'),es = require('event-stream');

var async = require('async');

var fileRead = false;
var lineNr = 0;

var q = async.queue(function(task,callback) {
    task(function(err,lineData){
        responseLines.push(lineData);
        callback();
      });
  },5);

var q.drain = function() {
  if(fileRead){
    done(null,responseLines);
  }
}

var s = fs.createReadStream('very-large-file.csv')
    .pipe(es.split())
    .pipe(es.mapSync(function(line){
        s.pause();
        q.push(async.apply(insertIntoDb,line))
        s.resume();
    })
    .on('error',function(err){
       done(err);
    })
    .on('end',function(){
      fileRead = true;
    })
);

还是有更好的使用异步,这将允许我这样做？
异步处理一行一行,提前退出的能力,如果其中一行有错误

解决方法

首先,我不知道你的代码段有多少是伪代码,但是var q.drain = …是无效的JavaScript,应该是错误的.它应该是q.drain =,因为你在一个现有对象上定义一个属性,而不是声明一个新的变量.这可能是为什么如果不是伪代码,你的漏极功能不会触发.

有几种方法可以实现我想要做的事情.一个是在结束处理程序中检查队列的长度,如果还有待处理的项目,请设置排水功能.

.on('end',function(){
  if(!q.length){
    callDone();
  }
  else {
    q.drain = callDone;
  }
});

function callDone(){
  done(null,responseLines);
}

这是有效的说“如果队列被处理了调用完成,如果没有,调用完成了！我确定有很多方法来整理您的代码,但希望这样可以解决您的具体问题.