我们正在尝试在 Node 应用程序中使用 Azure 服务总线。我们的要求是从队列中获取多条消息.
由于Azure SDK for Node不支持批量检索，我们决定使用AMQP。虽然我们能够使用 Peek Messages 获取消息，如此处所述（https://learn.microsoft.com/en-us/azure/service-bus-messaging/service-bus-amqp-request-response#message-operations https://learn.microsoft.com/en-us/azure/service-bus-messaging/service-bus-amqp-request-response#message-operations).
我们注意到，一旦获取消息，它们就会从队列中删除。我想知道是否有人了解如何使用 AMQP 和 Node.js 在“PeekLock”模式下获取消息。对于 AMQP，我们使用 amqp10 节点包（https://www.npmjs.com/package/amqp10 https://www.npmjs.com/package/amqp10).
这是我们查看消息的代码：
const AMQPClient = require('amqp10/lib').Client,
Policy = require('amqp10/lib').Policy;
const protocol = 'amqps';
const keyName = 'RootManageSharedAccessKey';
const sasKey = 'My Shared Access Key'
const serviceBusHost = 'account-name.servicebus.windows.net';
const uri = protocol + '://' + encodeURIComponent(keyName) + ':' + encodeURIComponent(sasKey) + '@' + serviceBusHost;
const queueName = 'test1';
var client = new AMQPClient(Policy.ServiceBusQueue);
client.connect(uri)
.then(function () {
return Promise.all([
client.createReceiver(queueName),
client.createSender(queueName)
]);
})
.spread(function(receiver, sender) {
console.log(receiver);
console.log(sender);
console.log('--------------------------------------------------------------------------');
receiver.on('errorReceived', function(err) {
// check for errors
console.log(err);
});
receiver.on('message', function(message) {
console.log('Received message');
console.log(message);
console.log('------------------------------------');
});
return sender.send([], {
operation: 'com.microsoft:peek-message',
'message-count': 5
});
})
.error(function (e) {
console.warn('connection error: ', e);
});
默认情况下接收器工作在自动稳定模式，你必须将其更改为决定处置:
const { Constants } = require('amqp10')
//
// ...create client, connect, etc...
//
// Second parameter of createReceiver method enables overwriting policy parameters
const receiver = client.createReceiver(queueName, {
attach: {
rcvSettleMode: Constants.receiverSettleMode.settleOnDisposition
}
})
处理消息后不要忘记接受/拒绝/释放消息：
receiver.on('message', msg => {
//
// ...do something smart with a message...
//
receiver.accept(msg) // <- manually settle a message
})
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)
