尝试使用用 Node.js 编写的 WebJob 轮询 Azure 服务总线队列。我创建了 2 个 WebJobs。第一个是按需的,并向队列发送 10 条唯一的消息。第二个作业是连续的,并轮询队列中的消息。
遇到以下问题:
轮询速度很慢。平均需要10分钟左右才能收到10条消息。请参阅下面的示例日志详细信息。这个速度基本没法用。所有的延迟都是因为得到回复receiveQueueMessage
。响应时间从 0 秒到约 120 秒不等,平均为 60 秒。
消息以随机顺序接收。不是先进先出。
有时消息会被接收两次,即使它们是在 ReceiveAndDelete 模式下读取的(我尝试过不使用读取模式参数,该参数应默认为 ReceiveAndDelete,使用{isReceiveAndDelete:true}
与{isPeekLock:false}
具有相同的结果)。
当队列为空时,它应该保持接收请求打开一天,但它总是在 230 秒后返回,并显示无消息错误。根据文档,最长为 24 天,所以我不知道 230 秒从何而来:
服务总线中阻塞接收操作的最大超时
排队时间是24天。但是,基于 REST 的超时有一个最大值
55秒。
基本上没有什么像宣传的那样有效。我究竟做错了什么?
发送消息测试作业:
var uuid = require('node-uuid');
var azure = require('azure');
var serviceBus = azure.createServiceBusService(process.env.busSearchConnectionString);
var messagesToSend = 10;
sendMessage(0);
function sendMessage(count)
{
var message = {
body: 'test message',
customProperties: {
message_number: count,
sent_date: new Date
},
brokerProperties: {
MessageId: uuid.v4() //ensure that service bus doesn't think this is a duplicate message
}
};
serviceBus.sendQueueMessage(process.env.busSearchQueueName, message, function(err) {
if (!err) {
console.log('sent test message number ' + count.toString());
} else {
console.error('error sending message: ' + err);
}
});
//wait 5 seconds to ensure messages are received by service bus in correct order
if (count < messagesToSend) {
setTimeout(function(newCount) {
//send next message
sendMessage(newCount);
}, 5000, count+1);
}
}
接收消息连续作业:
console.log('listener job started');
var azure = require('azure');
var serviceBus = azure.createServiceBusService(process.env.busSearchConnectionString);
listenForMessages(serviceBus);
function listenForMessages(serviceBus)
{
var start = process.hrtime();
var timeOut = 60*60*24; //long poll for 1 day
serviceBus.receiveQueueMessage(process.env.busSearchQueueName, {timeoutIntervalInS: timeOut, isReceiveAndDelete: true}, function(err, message) {
var end = process.hrtime(start);
console.log('received a response in %ds seconds', end[0]);
if (err) {
console.log('error requesting message: ' + err);
listenForMessages(serviceBus);
} else {
if (message !== null && typeof message === 'object' && 'customProperties' in message && 'message_number' in message.customProperties) {
console.log('received test message number ' + message.customProperties.message_number.toString());
listenForMessages(serviceBus);
} else {
console.log('invalid message received');
listenForMessages(serviceBus);
}
}
});
}
示例日志输出:
[05/06/2015 21:50:14 > 8c2504: SYS INFO] Status changed to Running
[05/06/2015 21:50:14 > 8c2504: INFO] listener job started
[05/06/2015 21:51:23 > 8c2504: INFO] received a response in 1s seconds
[05/06/2015 21:51:23 > 8c2504: INFO] received test message number 0
[05/06/2015 21:51:25 > 8c2504: INFO] received a response in 2s seconds
[05/06/2015 21:51:26 > 8c2504: INFO] received test message number 4
[05/06/2015 21:51:27 > 8c2504: INFO] received a response in 1s seconds
[05/06/2015 21:51:27 > 8c2504: INFO] received test message number 7
[05/06/2015 21:51:28 > 8c2504: INFO] received a response in 0s seconds
[05/06/2015 21:51:29 > 8c2504: INFO] received test message number 9
[05/06/2015 21:51:49 > 8c2504: INFO] received a response in 20s seconds
[05/06/2015 21:51:49 > 8c2504: INFO] received test message number 1
[05/06/2015 21:53:35 > 8c2504: INFO] received a response in 106s seconds
[05/06/2015 21:53:35 > 8c2504: INFO] received test message number 1
[05/06/2015 21:54:26 > 8c2504: INFO] received a response in 50s seconds
[05/06/2015 21:54:26 > 8c2504: INFO] received test message number 5
[05/06/2015 21:54:35 > 8c2504: INFO] received a response in 9s seconds
[05/06/2015 21:54:35 > 8c2504: INFO] received test message number 9
[05/06/2015 21:55:28 > 8c2504: INFO] received a response in 53s seconds
[05/06/2015 21:55:28 > 8c2504: INFO] received test message number 2
[05/06/2015 21:57:26 > 8c2504: INFO] received a response in 118s seconds
[05/06/2015 21:57:26 > 8c2504: INFO] received test message number 6
[05/06/2015 21:58:28 > 8c2504: INFO] received a response in 61s seconds
[05/06/2015 21:58:28 > 8c2504: INFO] received test message number 8
[05/06/2015 22:00:35 > 8c2504: INFO] received a response in 126s seconds
[05/06/2015 22:00:35 > 8c2504: INFO] received test message number 3
[05/06/2015 22:04:25 > 8c2504: INFO] received a response in 230s seconds
[05/06/2015 22:04:25 > 8c2504: INFO] error requesting message: No messages to receive
[05/06/2015 22:08:16 > 8c2504: INFO] received a response in 230s seconds
[05/06/2015 22:04:25 > 8c2504: INFO] error requesting message: No messages to receive