SubscriptionClient.RecieveBatch 未检索所有代理消息答案

【问题标题】：SubscriptionClient.RecieveBatch not retrieving all the brokered messagesSubscriptionClient.RecieveBatch 未检索所有代理消息
【发布时间】：2015-10-19 08:03:16
【问题描述】：

我有一个控制台应用程序来读取 Azure 服务总线上订阅中存在的所有代理消息。我那里有大约 3500 条消息。这是我阅读消息的代码：

SubscriptionClient client = messagingFactory.CreateSubscriptionClient(topic, subscription);   
long count = namespaceManager.GetSubscription(topic, subscription).MessageCountDetails.ActiveMessageCount;
Console.WriteLine("Total messages to process : {0}", count.ToString()); //Here the number is showing correctly
IEnumerable<BrokeredMessage> dlIE = null;
dlIE = client.ReceiveBatch(Convert.ToInt32(count));

当我执行代码时，在 dlIE 中，我只能看到 256 条消息。我也尝试过像 client.PrefetchCount 这样给出预取计数，但它也只返回 256 条消息。

我认为一次可以检索的消息数量有一些限制。但是在RecieveBatch方法的msdn页面上没有提到这样的事情。如何一次检索所有消息？

注意：

我只想读取消息，然后让它存在于服务总线上。因此我不使用message.complete 方法。
我无法从服务总线中删除并重新创建主题/订阅。

编辑：

我像这样使用 PeekBatch 而不是 ReceiveBatch：

    IEnumerable<BrokeredMessage> dlIE = null;
                            List<BrokeredMessage> bmList = new List<BrokeredMessage>();
  long i = 0;
   dlIE = subsciptionClient.PeekBatch(Convert.ToInt32(count)); // count is the total number of messages in the subscription.
  bmList.AddRange(dlIE);
  i = dlIE.Count();
 if(i < count)
  {           
 while(i < count)
  {
  IEnumerable<BrokeredMessage> dlTemp = null;
   dlTemp = subsciptionClient.PeekBatch(i, Convert.ToInt32(count));
    bmList.AddRange(dlTemp);
    i = i + dlTemp.Count();
    }
    }

我的订阅中有 3255 条消息。当第一次调用 peekBatch 时，它会收到 250 条消息。所以它使用PeekBatch(250,3225) 进入while 循环。每次只收到 250 条消息。我在输出列表中的最终总消息是 3500 重复。我无法理解这是如何发生的。

【问题讨论】：

标签： c# azure azureservicebus azure-servicebus-queues azure-servicebus-topics

【解决方案1】：

我想通了。订阅客户端会记住它检索到的最后一批，并在再次调用时检索下一批。

所以代码是：

    IEnumerable<BrokeredMessage> dlIE = null;
List<BrokeredMessage> bmList = new List<BrokeredMessage>();
  long i = 0;  
  while (i < count)
  {
   dlIE = subsciptionClient.PeekBatch(Convert.ToInt32(count));
   bmList.AddRange(dlIE);
   i = i + dlIE.Count();
  }

感谢 MikeWo 的指导

注意：您一次可以查看的消息数量似乎有某种大小限制。我尝试了不同的订阅，并且每个订阅的消息数量都不同。

【讨论】：

我在回答中提到了这一点。 “通过在引擎盖下使用与 PeekBatch 相同的 SubscriptionClient，最后拉出的序列号会在您循环时保留，因为它应该跟踪并遍历整个队列。”
@MikeWo 抱歉，我没有在此处阅读您的答案，而是针对 SO 中的其他问题。实际上，在我在这里看到您的答案之前，我正在根据您在其他地方的答案使用 peekbatch。
@MikeWo PeekBatch 检索的消息数量是否有限制？它适用于 3200，但是当订阅上有 5000 条消息时，它会显示此错误：内部服务器错误：服务器没有提供有意义的回复；这可能是由于会话过早关闭造成的。
@MikeWo 我认为有一些主题问题。因为即使在其中给出 peekbatch(250) 也会产生相同的错误。包含 3200 条消息的订阅（有效）在另一个主题中。
理论上我能找到的文档中没有提到最大值，并且 messageCount 参数是一个 Int；但是，我认为我永远不会指望能够像那样一次抓住非常大的群体。偷看可能没那么糟糕，但这取决于它们有多大、大小的一致性以及你对它们的一般处理方式。至于错误，就像您说的那样，这可能是其他问题，而不是消息计数的问题。

【解决方案2】：

您正在写的主题是偶然分区的吗？当您从分区实体接收消息时，它一次只会从其中一个分区中获取。 From MSDN:

“当客户端想要从分区队列或分区主题的订阅接收消息时，服务总线会查询所有片段以获取消息，然后返回从任何消息传递返回的第一条消息存储到接收者。服务总线缓存其他消息并在收到其他接收请求时返回它们。接收客户端不知道分区；分区队列或主题的面向客户端的行为（例如，读取、完成、 defer, deadletter, prefetching) 与常规实体的行为相同。”

假设即使使用非分区实体，您也可以使用 Receive 或 Peek 方法一次性获取所有消息，这可能不是一个好主意。以更小的批次循环遍历消息会更有效率，特别是如果您的消息对它们有任何合适的大小或大小不确定。

由于您实际上并不想从队列中删除消息，我建议使用PeekBatch 而不是 ReceiveBatch。这使您可以获取消息的副本并且不会锁定它。我强烈建议将相同的 SubscriptionClient 与 PeekBatch 结合使用的循环。通过在引擎盖下使用与 PeekBatch 相同的 SubscriptionClient，最后拉出的序列号被保留，因为您循环遍历它应该跟踪并遍历整个队列。这基本上可以让您阅读整个队列。

【讨论】：

您好，感谢您的回复。我正在使用 PeekBatch 方法，但遇到了一些问题。我已经更新了这个问题。你能看一下吗？

【解决方案3】：

我遇到了一个类似的问题，client.ReceiveBatchAsync(....) 不会从 azure 服务总线中的订阅中检索任何数据。

经过一番挖掘，我发现每个订阅者都有一点可以启用批处理操作。这只能通过 powershell 启用。以下是我使用的命令：

$subObject = Get-AzureRmServiceBusSubscription -ResourceGroup '#resourceName' -NamespaceName '#namespaceName' -Topic '#topicName' -SubscriptionName '#subscriptionName' 
$subObject.EnableBatchedOperations = $True

Set-AzureRmServiceBusSubscription -ResourceGroup '#resourceName' -NamespaceName '#namespaceName' -Topic '#topicName'-SubscriptionObj $subObject

更多详情请见here。虽然它仍然没有加载所有消息，但至少它开始清除队列。据我所知，批量大小参数只是作为对服务总线的建议而不是规则。

希望对你有帮助！

【讨论】：