如何使用AMQP在"PeekLock"模式下从Azure服务总线队列获取消息？

Question

我们试图在节点应用程序中使用Azure服务总线。我们的要求是从一个队列中获取多个消息。

由于Azure SDK for Node不支持批量检索，我们决定使用AMQP。而我们可以使用这里描述的Peek消息获取消息(https://docs.microsoft.com/en-us/azure/service-bus-messaging/service-bus-amqp-request-response#message-操作）。

我们注意到的是，一旦获取消息，它们就会从队列中删除。我想知道，如果我在使用AMQPAM模式时使用AMQPAM来获取消息，我将如何锁定任何人。对于AMQP，我们使用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);
});

Answer 1

默认情况下，接收方在自动结算模式下工作，您必须将其更改为处置结算：

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
})