c# - Azure 服务总线 SubscriptionClient 高延迟/无法同时接收消息

Question

如果我向某个主题发送一批消息，并使用订阅客户端读取消息，那么我似乎会按顺序接收消息，即每发送一条消息都会触发 OnMessageAsync，但是有一个每个接收事件之间有明显的(150+ 毫秒)延迟

发件人:

var factory = MessagingFactory.CreateFromConnectionString("blah");
sender = factory.CreateMessageSender("MyTopicName");

var tasks = new List<Task>();
for (int i = 0; i < 10; i++)
    tasks.Add(sender.SendAsync(new BrokeredMessage("My Message"))
       .ContinueWith(t => Log("Sent Message {i}"));

await Task.WhenAll(tasks);  // This completes within a few millis

接收者:

receiver = factory.CreateSubscriptionClient("MyTopicName", "MySubscription");
_sbClient.OnMessageAsync(async message =>
{
    var msg = message.GetBody<string>();
    Log("Received message xxxx
    await message.CompleteAsync();
});

这意味着发送的第 10 条消息仅在发送后超过 1.5 秒才被接收。

Azure 延迟测试显示我正在使用的数据中心有大约 200 毫秒的延迟，因此我不希望消息在此之前返回(实际上第一条消息是在这之后不久收到的)，但我不会期待我所看到的“累积”行为。

使用 MaxConcurrentCalls 并在 OnMessageAsync 中添加延迟，显示此功能按预期工作，并且我可以看到一次仅处理 MaxConcurrentCalls

我搞乱了DeleteOnReceive模式，启用“Express”，禁用“分区”，使用AMQP strong> 而不是 SBMP 等，但是似乎没有什么真正有太大区别。

[我正在使用 Microsoft.ServiceBus，版本=3.0.0.0]

编辑:
日志如下所示。因此，如果我同时发送 10 条消息，我只会在发送后 1.5 秒收到第 10 条消息:

18:09:32.624 Sent message 0
18:09:32.624 Sent message 1
18:09:32.641 Sent message 2
18:09:32.641 Sent message 3
18:09:32.674 Sent message 4
18:09:32.674 Sent message 5
18:09:32.709 Sent message 6
18:09:32.709 Sent message 7
18:09:32.738 Sent message 8
18:09:32.738 Sent message 9

18:09:32.791 Received message 1 in 341 millis
18:09:32.950 Received message 2 in 487 millis
18:09:33.108 Received message 3 in 628 millis
18:09:33.265 Received message 4 in 770 millis
18:09:33.426 Received message 5 in 914 millis
18:09:33.586 Received message 6 in 1060 millis
18:09:33.745 Received message 7 in 1202 millis
18:09:33.906 Received message 8 in 1347 millis
18:09:34.065 Received message 9 in 1492 millis

Answer 1

在深入研究了 OnMessage 消息泵的工作原理之后，我意识到这实际上是一种轮询机制，其中对 ServiceBus 的底层调用仍然是“Receive()' 尝试提取任何新消息。如果超时，则调用会无限次地再次进行。如果对 Receive() 的调用仅返回一条消息，然后需要 150 毫秒的往返时间来检索下一条消息等，那么我看到的行为是有意义的。

输入PrefetchCount 。在 SubscriptionClient 上将其设置为非零值可以有效地允许底层 Receive() 拉取多个消息，然后将这些消息缓存起来并(立即)可用于冒泡到 OnMessage 中。