Node js - 如何处理多个异步任务

Question

我是 node 新手，在处理多个异步任务时遇到了困难。

除了节点，我还有另一个服务器（S1），它不会立即向请求返回数据，它可以返回多种类型的数据，也可以发送通知而不需要专门请求它们，所以节点必须监听数据从它，解析它并采取相应的行动。

与此服务器 (S1) 的连接通过以下方式完成：

S1 = net.createConnection({'host':S1Host, 'port': S1Port});

节点通过以下方式监听数据：

S1.on('data', function(data){
  S1DataParse(data);
});

我必须将正确的数据（在解析后）路由到特定的 POST 请求。

app.post('/GetFooFromS1', function(req, res){

  // Send request to S1
  S1.write({'type':'foo'});

  // If got the correct data sometime in the future, send response to the browser
  res.setHeader('Content-Type', 'application/json');
  res.json({'status':'success', 'value':S1FooData});
});

我尝试为此使用异步模块，但没有成功。 我想做什么：

var asyncTasks = [];

app.post('/GetFooFromS1', function(req, res){

  asyncTasks.push(function(callback){
    // Send request to S1
    S1.write({'type':'foo'});
  });

  async.parallel(asyncTasks, function(response){
    res.setHeader('Content-Type', 'application/json');
    res.json({'status':'success', 'value':response});
  });
});

S1DataParse 中的另一个任务：

function S1DataParse(){
  if(data.type='foo'){
    asyncTasks.push(function(callback){
      callback(data);
    });
  }
}

但是，当然，第二个任务从未添加到 asyncTasks 数组中。我真的被困住了。 你能帮帮我吗？

谢谢

-=-=-=- 编辑 -=-=-=-

最后，我遇到了 events 和 EventEmitter()。

从 POST 请求中，我调用向数据服务器发送请求的函数 (DataServerClientGet)。 在这个函数中，我注册了一个监听器，它将获取未来的数据。 eventEmitter.on('getData', returnDataServerData);

这一切都很好，除了一件事。每当我刷新页面或添加其他 POST 请求时，都会出现错误：

错误：发送后无法设置标头。

如果我能解决这个问题，那就太好了。请帮帮我。

谢谢 ;)

整个代码如下所示：

var express = require('express');
var app = express();
var http = require('http').Server(app);
var bodyParser = require('body-parser')
var net = require('net');
var events = require('events');

var dataServerHost = '127.0.0.1';
var dataServerPort = 12345;
var dataServerClient;
var logMsg;

var eventEmitter = new events.EventEmitter();


/*******************************************/
//                Init
/*******************************************/
app.use(bodyParser.json());
app.use(bodyParser.urlencoded({extended: true}));
app.use(express.static(__dirname + '/public'));


/*******************************************/
//       Connect to the data server
/*******************************************/
DataServerConnect();


/*******************************************/
// Open listener on port 3000 (to browser)
/*******************************************/
http.listen(3000, function(){
  logMsg = 'listening on *:3000';
  console.log(logMsg);
});


/*******************************************/
//                Routing
/*******************************************/
app.get('/', function(req, res){
  res.sendFile(__dirname + '/index.html');
});

app.post('/GetDataFoo', function(req, res){
  var msg;
  var size;

  msg ='\n{"Type":"Query", "SubType":"GetDataFoo","SearchFilter":""}';
  size = msg.length;

  logMsg = 'Client to DataServer: GetDataFoo';
  console.log(logMsg);

  DataServerClientGet('GetDataFoo', size, msg, res);
});


/*******************************************/
//               Functions
/*******************************************/
function DataServerConnect(){
  dataServerClient = net.createConnection({'host':dataServerHost, 'port': dataServerPort}, function(){
    logMsg = 'Connected to DataServer ['+dataServerHost+':'+dataServerPort+']';
    console.log(logMsg);
  });

  dataServerClient.on('data', function(data){

    logMsg = 'DataServerData>>>\n'+data.toString()+'DataServerData<<<';
    console.log(logMsg);

    DataServerDataParse(data.toString());
  });

  dataServerClient.on('end', function(){
    logMsg = 'Disconnected from DataServer';
    console.log(logMsg);
  });
}

function DataServerClientGet(type, size, msg, res){
  dataServerClient.write('Type: Json\nSize: '+size+'\n\n'+msg, function(err){

  var returnDataServerData = function returnDataServerData(results){
    res.setHeader('Content-Type', 'application/json');
    res.json({'status':'success', 'value':results});
  }
  eventEmitter.on('getData', returnDataServerData);
}

function DataServerDataParse(json){
  if(json.Type=='GetDataFoo')
  {
    var MessageList = json.MessageList;
    eventEmitter.emit('getData', MessageList);
  }
}

-=-=-=- 编辑 -=-=-=-

Error: Can't set headers after they are sent. 是由于每次调用 DataServerClientGet 时添加相同类型的相同侦听器而导致的发送多次。

我通过添加解决了这个问题：removeListener(event, listener) 在 res 之后，在函数内部。无论如何，我认为这是错误的，如果多次调用具有相同类型的 DataServerClientGet 等，可能会导致问题。

Answer 1

有一个可选的回调参数可以传递给写函数（docs），类似于：

S1.write({'type':'foo'},function(err){
  if(err){
   //Handle error
  }else{
     res.setHeader('Content-Type', 'application/json');
     res.json({'status':'success', 'value':response});
  }
})

这可以与发布路由一起使用，但是在您的“数据”侦听器中，如果您想要双向行为，您可以检查@987654322 当客户端没有初始化连接时（它不是双向），您不能将数据从服务器发送到客户端@