Azure Service Bus会话闲置超时后无消息事件处理方案问询

你的需求完全可行，Azure Service Bus的SessionIdleTimeout就是为处理会话闲置场景设计的——当会话在设定的15秒内未收到新消息、也无其他交互操作时，会触发会话关闭逻辑，我们可以利用这个时机完成数据库更新和停止处理的操作。

核心实现思路 #

ServiceBusSessionProcessor提供了ProcessSessionClosingAsync事件，该事件会在会话关闭时触发，其中就包含闲置超时导致的关闭场景。我们可以通过事件参数的Reason属性判断关闭原因，再执行对应的业务逻辑。

修改后的代码示例 #

public BaseSessionMessageProcessor(
    IMessageHandler<T> handler,
    string topicName,
    string subscriptionName,
    string sessionId,
    IServiceBusPersisterConnection serviceBusPersisterConnection,
    IMessageHelper messageHelper,
    IAppLogger<T> logger)
{
    _handler = handler;
    _messageHelper = messageHelper;
    _logger = logger;

    var options = new ServiceBusSessionProcessorOptions
    {
        MaxConcurrentSessions = 2,
        MaxConcurrentCallsPerSession = 1,
        AutoCompleteMessages = false,
        SessionIds = { sessionId },
        SessionIdleTimeout = TimeSpan.FromSeconds(15)
    };
    _processor = serviceBusPersisterConnection.ServiceBusClient.CreateSessionProcessor(topicName, subscriptionName, options);

    RegisterSessionMessageProcessor().GetAwaiter().GetResult();
}

public async Task RegisterSessionMessageProcessor()
{
    _processor.ProcessMessageAsync += SessionMessageHandler;
    _processor.ProcessErrorAsync += ErrorHandler;
    // 订阅会话关闭事件
    _processor.ProcessSessionClosingAsync += SessionClosingHandler;

    await _processor.StartProcessingAsync();
}

public async Task SessionMessageHandler(ProcessSessionMessageEventArgs args)
{
    _logger.LogInformation($"log-info: process started");
    // 处理消息的核心逻辑...
    // 消息处理完成后需调用CompleteAsync确认，否则会影响会话闲置计时
    await args.CompleteMessageAsync(args.Message);
}

public Task ErrorHandler(ProcessErrorEventArgs args)
{
    var exception = args.Exception;
    var context = args.ErrorSource;

    _logger.LogError($"log-exception: An error occurred while trying to handle a message.", exception, context);

    return Task.CompletedTask;
}

// 会话关闭事件处理方法
public async Task SessionClosingHandler(ProcessSessionClosingEventArgs args)
{
    // 精准判断是否为闲置超时导致的会话关闭
    if (args.Reason == SessionCloseReason.IdleTimeout)
    {
        _logger.LogInformation($"log-info: Session {args.SessionId} closed due to idle timeout.");
        
        // 执行数据库更新操作
        await UpdateSessionStatusInDb(args.SessionId);
        
        // 停止处理器接收新消息
        await _processor.StopProcessingAsync();
    }
}

// 数据库更新逻辑（示例）
private async Task UpdateSessionStatusInDb(string sessionId)
{
    // 替换为你的实际数据库操作
    // 示例：
    // var sessionRecord = await _dbContext.SessionTrackers.FirstOrDefaultAsync(s => s.SessionId == sessionId);
    // if (sessionRecord != null)
    // {
    //     sessionRecord.Status = "IdleTimeoutCompleted";
    //     await _dbContext.SaveChangesAsync();
    // }
}

关键细节说明 #

• SessionCloseReason.IdleTimeout：该枚举值精准标识了会话因闲置超时关闭的场景，确保我们只在符合需求的时机执行逻辑。
• 消息处理确认：在SessionMessageHandler中必须调用CompleteAsync（或AbandonAsync等）完成消息处理，否则未确认的消息会影响会话的闲置计时逻辑。
• 停止处理器：StopProcessingAsync是异步操作，需等待执行完成，确保处理器彻底停止接收新的会话和消息。

内容的提问来源于stack exchange，提问作者alex