从 Azure 服务总线读取延迟消息时会话锁定已过期答案

【问题标题】：Session lock expired on reading deferred message from Azure Service Bus从 Azure 服务总线读取延迟消息时会话锁定已过期
【发布时间】：2022-10-08 18:08:04
【问题描述】：

我有 2 个应用程序，其中应用程序 A 正在向应用程序 B 发送复杂对象的数据集。必须按照每个数据集的特定顺序处理对象，这就是为什么我使用在 Azure 服务总线上启用会话的主题，其中每条消息都有一个会话 ID（代表一个数据集）和一个代表顺序的序列 ID。

我正在使用ServiceBusSessionProcessor 从总线接收消息，它的初始化如下：

SessionOptions = new ServiceBusSessionProcessorOptions()
{
   MaxAutoLockRenewalDuration = TimeSpan.FromMinutes(5),
   AutoCompleteMessages = false,
   MaxConcurrentSessions = 4,
   ReceiveMode = ServiceBusReceiveMode.PeekLock,
   SessionIdleTimeout = TimeSpan.FromMinutes(10),
   MaxConcurrentCallsPerSession = 1,
};
SubscriptionSessionProcessor = SubscriptionClient.CreateSessionProcessor("TestTopic", "TestSubscription", SessionOptions);

收到消息后，我检查消息是否在要处理的序列中的下一个，如果是，则处理消息，否则我将消息推迟到总线上

protected async Task HandleSessionMessageAsync<R>(SessionInformationBase sessionInformation, ServiceBusReceivedMessage message)
{
   var stateData = await sessionInformation.Message.GetSessionStateAsync();
   var sessionState = SessionStateManager.CreateSessionStateManager(stateData);

   if (sessionState.IsNextMessage(sessionInformation.SessionData.Sequence))
   {
      try
      {
         sessionInformation.ProcessorCall.Invoke();
      }
      catch (Exception e)
      {
         ExceptionTracer.TraceException(e);
      } 
      await sessionInformation.Message.CompleteMessageAsync(message);
      await HandleSessionStateAsync<R>(sessionInformation, sessionState); 
   }
   else
   {
      sessionState.DeferredMessages.Add(sessionInformation.SessionData.Sequence, message.SequenceNumber);
      //defer message directly on the message object (the message that was received on the ServiceBusSessionProcessor object
      await sessionInformation.Message.DeferMessageAsync(message);  
      await SetSessionStateAsync(sessionInformation.Message, sessionState.Serialize());
   }
}

当我按顺序处理一条消息时，我检查是否已经收到了下一条消息，如果是，我尝试从总线上读取它，如下所示：

if (!sessionState.NextMessageIsDeferred())
   return;
try
{
   var deferredMessage = await Receiver.ReceiveDeferredMessageAsync(sessionState.GetNextMessageSequenceId());
   ... //process deferred message
}

但是，当我调用ReceiveDeferredMessageAsync 方法时，我收到一个会话锁过期异常。会话锁定超时设置为 5 分钟，我知道延迟和读取延迟消息之间只有几秒钟。

一个潜在的问题是我正在使用ServiceBusReceiver 类的对象来检索延迟消息，它既不是接收消息的原始处理器也不是对象本身 - 但据我所知，这些对象无法检索延迟消息。

我尝试通过接收器对象延迟消息，用接收器对象替换处理器对象，但我总是遇到不同的死胡同。

我为此使用 Azure.Messaging.ServiceBus 7.10.0 库。

我缺少什么想法-为什么我的会话锁定会过期？

【问题讨论】：

标签： .net azureservicebus

【解决方案1】：

所以设法弄清楚了这一点。根据文档，延迟消息只能由首先延迟它的生产者检索。在我的示例中，情况并非如此，但我重构了我的代码以使用 ServiceBusReceiver 从总线检索消息，如果消息不是序列中的下一个消息，则接收者推迟消息并且同一个接收者是能够再次检索延迟的消息。

【讨论】：