如何在处理单条消息后释放Azure Service Bus会话锁？

使用启用会话的Service Bus主题订阅，期望处理完单条消息后立即释放会话锁——即第一个ServiceBusSessionProcessor调用CompleteMessageAsync后，第二个ServiceBusSessionProcessor能立刻获取同一sessionId的消息。但当前需等待5分钟锁过期后，第二个处理器才能获取消息。

现有代码 #

创建会话处理器的调用代码 #

private void CreateSessionMessageProcessor(string sessionId)
{
    IMessageProcessorFactory factory = new MessageProcessorFactory(_serviceProvider.GetRequiredService<IServiceBusPersisterConnection>());

    var serviceScope = _serviceProvider.CreateScope();

    factory.CreateSessionMessageProcessor(
    serviceScope.ServiceProvider.GetRequiredService<IMessageHandler<Message>>(),
    "topic",
    "subscription",
    sessionId,
    TimeSpan.FromSeconds(15),
    _serviceProvider.GetRequiredService<IMessageHelper>(),
    serviceScope.ServiceProvider.GetRequiredService<IAppLogger<Message>>());
}

会话处理器实现代码 #

public ISessionMessageProcessor CreateSessionMessageProcessor<T>(IMessageHandler<T> handler, string topicName, string subscriptionName, string sessionId, TimeSpan timeout, IMessageHelper messageHelper, IAppLogger<T> logger)
{
    return new BaseSessionMessageProcessor<T>(handler, topicName, subscriptionName, sessionId, timeout, _serviceBusPersisterConnection, messageHelper, logger);
}


public BaseSessionMessageProcessor(
    IMessageHandler<T> handler,
    string topicName,
    string subscriptionName,
    string sessionId,
    TimeSpan timeOut,
    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 = timeOut,
        MaxAutoLockRenewalDuration = TimeSpan.Zero
    };
    _processor = serviceBusPersisterConnection.ServiceBusClient.CreateSessionProcessor(topicName, subscriptionName, options);

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

public async Task RegisterSessionMessageProcessor()
{
    _processor.ProcessMessageAsync += SessionMessageHandler;
    _processor.ProcessErrorAsync += ErrorHandler;

    await _processor.StartProcessingAsync();
}

public async Task SessionMessageHandler(ProcessSessionMessageEventArgs args)
{
    _logger.LogInformation($"log-info: process started");
}

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;
}

解决方案 #

核心问题在于会话处理器默认会保持会话锁定，直到SessionIdleTimeout触发或处理器停止。要实现处理完单条消息立即释放锁，需主动关闭会话：

1. 修改消息处理方法：在完成消息后调用args.Session.CloseAsync()，主动释放会话锁。修改后的SessionMessageHandler如下：

public async Task SessionMessageHandler(ProcessSessionMessageEventArgs args)
{
    _logger.LogInformation("log-info: process started");
    try
    {
        // 执行消息处理逻辑
        await _handler.HandleAsync(args.Message);
        
        // 标记消息为已处理
        await args.CompleteMessageAsync(args.Message);
        
        // 主动关闭会话，立即释放会话锁
        await args.Session.CloseAsync();
    }
    catch (Exception ex)
    {
        _logger.LogError(ex, "处理消息时发生异常");
        // 根据业务需求处理失败消息，例如放弃消息
        await args.AbandonMessageAsync(args.Message);
    }
}

2. 配置验证：确保ServiceBusSessionProcessorOptions中的MaxAutoLockRenewalDuration保持为TimeSpan.Zero，避免自动续期会话锁；SessionIdleTimeout可保留现有配置，作为会话空闲时的兜底释放机制。

3. 处理器竞争逻辑：当第一个处理器关闭会话后，同一sessionId的其他ServiceBusSessionProcessor会立即竞争获取会话锁，无需等待锁过期。

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