使用 Node JS 库的 Azure 服务总线队列的最大吞吐量？

Question

使用 Azure NodeJS 库，我无法以每秒 5 或 6 条消息的速度从 Azure 服务总线队列接收消息。这比官方文档建议的要慢几个数量级。我正在使用一个关闭分区的队列（这里推荐Polling an Azure Service Bus Queue from an Azure WebJob using Node.js），在ReceiveAndDelete 模式下读取。本质上我只是在重复调用 .receiveQueueMessage()。

基于这个问题 (Azure Service Bus Scalability) 似乎其他人在使用 NodeJS 时也每秒看到 5/6 条消息。这几乎是一个硬限制吗？任何已知的解决方法或优化？

Answer 1

您提供的线程中的代码很棒，它似乎在管道工作流中从服务总线队列接收消息，接收完上一条消息后将接收下一条消息。

正如official site 在场景阶段的高吞吐量队列部分中提到的，我们可以找到以下适合您情况的点：

• 要提高队列的整体接收速率，请使用多个 用于创建接收器的消息工厂。

• 使用异步操作来利用客户端 批处理。

• 将批处理间隔设置为 50ms 以减少 Service Bus 的数量 客户端协议传输。如果使用多个发件人，请增加 批处理间隔为 100ms。

为了实现这些优化，我们可以利用sample on Azure github repo。

并最大化单个队列的吞吐量。我利用上面的示例代码进行了一个简单的测试，并将循环时间设置为 10 毫秒。 我的测试结果是：在ReceiveAndDelete模式下每秒会收到近50条消息；在 PeekLock 模式下每秒会收到大约 70 条消息，显示在https://azure.microsoft.com/en-us/documentation/articles/service-bus-nodejs-how-to-use-queues/#receive-messages-from-a-queue

var uuid = require('node-uuid');
var azure = require('azure');
var serviceBus = azure.createServiceBusService(connectionString);

function checkForMessages(sbService, queueName, callback) {
  sbService.receiveQueueMessage(queueName,{ isPeekLock: true }, function (err, lockedMessage) {

    if (err) {
      if (err === 'No messages to receive') {
        console.log('No messages');
      } else {
        callback(err);
      }
    } else {
      callback(null, lockedMessage);
    }
  });
}

function processMessage(sbService, err, lockedMsg) {
  if (err) {
    console.log('Error on Rx: ', err);
  } else {
    console.log('Rx: ', lockedMsg);
    sbService.deleteMessage(lockedMsg, function(err2) {
      if (err2) {
        console.log('Failed to delete message: ', err2);
      } else {
        console.log('Deleted message.');
      }
    })
  }
}

var idx = 0;
function sendMessages(serviceBus, queueName) {
  var msg = 'Message # ' + (++idx) + (' '+uuid.v4());
  serviceBus.sendQueueMessage(queueName, msg, function (err) {
   if (err) {
     console.log('Failed Tx: ', err);
   } else {
     console.log('Sent ' + msg);
   }
  });
}


var queueName = 'myqueue';
serviceBus.getQueue(queueName, function (err,res) {
  if (err) {
   console.log('Failed: ', err);
  } else {
  console.log('current msg count '+ res.MessageCount);
   // var t = setInterval(checkForMessages.bind(null, serviceBus, queueName,function(err, lockedMsg){}), 10);  //ReceiveAndDelete mode
   var t = setInterval(checkForMessages.bind(null, serviceBus, queueName, processMessage.bind(null, serviceBus)), 10);  // PeekLock mode
   // setInterval(sendMessages.bind(null, serviceBus, queueName), 100);
                setTimeout(function(){
                                clearInterval(t);
                                console.log('task over');
                   },1000);
  }
});