Azure:具有唯一键的循环(环)队列

Question

对于我的应用程序，我需要组织一个循环(环形)队列。这意味着任何已处理的消息都会立即进入队列末尾继续处理。

例如:

1. 队列:A、B、C。
2. 接收方处理 A。
3. 队列:B、C、A。

2 和 3 应以原子方式执行。因此我们永远不会丢失 A 或任何其他消息。

另一个要求是忽略重复项。因此队列中应该始终有一个A。即使发送者推送另一个 A 项目。 A 在这里指的是消息的一些唯一(主)键。

我寻找使用 Azure 服务总线，但我找不到如何用它来满足这两个要求。是否可以通过 Service Bus 来实现该场景？如果没有，最好的替代方案是什么？

Answer 1

这种队列可以通过服务总线 session 来实现。 session 提供“分组依据”机制，因此我们可以将唯一键分配给消息的 SessionId，然后按组接收消息，忽略组中除第一个消息之外的所有消息。

实现

1) 创建一个队列，并将 RequiresSession 设置为 true:

var queueDescription = new QueueDescription("CircularQueue")
{
    RequiresSession = true,
};
await namespaceManager.CreateQueueAsync(queueDescription);

2)向队列发送消息时，将SessionId设置为您的唯一键值:

var message = new BrokeredMessage($"Message body")
{
    MessageId = "MESSAGE_UNIQUE_KEY",
    SessionId = "MESSAGE_UNIQUE_KEY"
};
await queueClient.SendAsync(message);

3)使用 session 接收消息:

while (true)
{
    var session = await queueClient.AcceptMessageSessionAsync(TimeSpan.FromSeconds(10));
    if (session == null)
        continue;

    try
    {
        var messages = (await session.ReceiveBatchAsync(100)).ToList();

        if (messages.Count == 0)
            continue;

        var message = messages[0];

        ProcessMessage(message);

        await queueClient.SendAsync(message.Clone());
        await session.CompleteBatchAsync(messages.Select(msg => msg.LockToken));
    }
    finally
    {
        await session.CloseAsync();
    }
}