处理消息队列并使用异步

Question

我编写了一个小型测试节点应用程序，它循环并将消息添加到队列（天蓝色存储队列），如下所示：

var queueService = azure.createQueueService();
var queueName = 'taskqueue';

// other stuff like check if created

// loop called after queue is confirmed

for (i=0;i<1000;i++){
  queueService.createMessage(queueName, "Hello world!", null, messageCreated);
}

// messageCreated does nothing at the moment, just logs to console

我正在尝试重写它来处理说 100 万个使用异步来控制并行运行的工作函数的数量的创建。这比什么都重要。

https://github.com/caolan/async#queue

这是异步队列的基本设置，我对需要更改的内容有点茫然。我认为以下方法行不通：

var q = async.queue(function (task, callback) {
   queueService.createMessage(queueName, task.msg, null, messageCreated);
    callback();
}, 100);


// assign a callback.  Called when all the queues have been processed
q.drain = function() {
    console.log('all items have been processed');
}

// add some items to the queue
for(i=0;i<1000000;i++) {
  q.push({msg: 'Hello World'}, function (err) {
    console.log('finished processing foo');
  });
    console.log('pushing: ' + i);
}

我不太明白如何将它们与异步结合在一起。

Answer 1

这是你的错误：

var q = async.queue(function (task, callback) {
   queueService.createMessage(queueName, task.msg, null, messageCreated);
    callback();
}, 100);

您正在做的是在队列上创建一条消息，然后立即传递延续（调用回调）。您要做的是在传递给createMessage 的回调中传递延续：

var q = async.queue(function (task, callback) {
   queueService.createMessage(queueName, task.msg, null, function(error, serverQueue, serverResponse) {
       callback(error, serverQueue, serverResponse);
       messageCreated(error, serverQueue, serverResponse);
   });
}, 100);

现在每个任务都会在实际创建任务后报告完成。

编辑：更新了createMessage 回调的接口。