Azure Service Bus同会话消息批量处理及失败回滚实现问询

核心思路 #

Azure Service Bus的会话机制本身不直接支持会话级原子事务，但可以通过会话状态跟踪 + 批量消息接收 + 自定义错误处理来实现需求：

• 利用SessionProcessor的会话生命周期事件管理整个会话的处理流程
• 在会话初始化时一次性接收该会话下的所有消息，确保处理的原子性
• 用会话状态记录处理进度，失败时回滚所有已处理的消息（放弃或重新入队）
• 复用SessionProcessor的重试、死信及会话切换能力

具体实现步骤 #

1. 配置SessionProcessor并绑定生命周期事件 #

修改原有的SessionProcessor初始化代码，重点关注会话生命周期事件的绑定：

public async Task StartAsync(CancellationToken cancellationToken)
{
    var processorOptions = new SessionProcessorOptions
    {
        MaxAutoLockRenewalDuration = TimeSpan.FromMinutes(5), // 延长锁时长适配批量处理
        SessionIdleTimeout = TimeSpan.FromMinutes(2),
        MaxConcurrentSessions = 5 // 根据服务器资源调整并发数
    };

    sessionProcessor = _serviceBusClient.CreateSessionProcessor(
        _serviceBusOptions.TopicName,
        _serviceBusOptions.SubscriptionName,
        processorOptions);

    // 绑定会话生命周期与错误处理事件
    sessionProcessor.SessionInitializingAsync += OnSessionInitializingAsync;
    sessionProcessor.SessionClosingAsync += OnSessionClosingAsync;
    sessionProcessor.ProcessErrorAsync += OnProcessErrorAsync;

    await sessionProcessor.StartProcessingAsync(cancellationToken);
}

2. 会话初始化时批量处理所有消息 #

在SessionInitializingAsync事件中，完成会话内所有消息的接收与原子性处理：

private async Task OnSessionInitializingAsync(SessionInitializingEventArgs args)
{
    var sessionReceiver = args.SessionReceiver;
    var sessionMessages = new List<ServiceBusReceivedMessage>();

    try
    {
        // 循环接收会话内所有消息，直到无新消息
        ServiceBusReceivedMessage message;
        do
        {
            message = await sessionReceiver.ReceiveMessageAsync(TimeSpan.FromSeconds(1));
            if (message != null)
            {
                sessionMessages.Add(message);
            }
        } while (message != null);

        if (!sessionMessages.Any()) return;

        // 标记会话为处理中
        await sessionReceiver.SetSessionStateAsync(BinaryData.FromObjectAsJson(new SessionState { Status = "Processing" }));

        // 用事务包裹业务处理与消息完成操作，确保原子性
        using var transaction = new TransactionScope(TransactionScopeAsyncFlowOption.Enabled);
        foreach (var msg in sessionMessages)
        {
            // 调用可能失败的业务服务
            await ProcessSessionRequest(msg);
        }

        // 所有处理成功，批量完成消息
        foreach (var msg in sessionMessages)
        {
            await sessionReceiver.CompleteMessageAsync(msg);
        }

        transaction.Complete();
        await sessionReceiver.SetSessionStateAsync(BinaryData.FromObjectAsJson(new SessionState { Status = "Completed" }));
    }
    catch (Exception ex)
    {
        // 处理失败，回滚所有消息（放弃后消息会重新入队）
        foreach (var msg in sessionMessages)
        {
            await sessionReceiver.AbandonMessageAsync(msg);
        }

        // 跟踪重试次数，超过阈值则将会话消息移入死信
        var sessionState = await sessionReceiver.GetSessionStateAsync();
        var currentState = sessionState != null 
            ? JsonSerializer.Deserialize<SessionState>(sessionState) 
            : new SessionState();
            
        currentState.RetryCount++;
        if (currentState.RetryCount >= 3)
        {
            foreach (var msg in sessionMessages)
            {
                await sessionReceiver.DeadLetterMessageAsync(msg, "Max retries reached", ex.Message);
            }
            currentState.Status = "DeadLetter";
        }
        else
        {
            currentState.Status = "Failed";
        }

        await sessionReceiver.SetSessionStateAsync(BinaryData.FromObjectAsJson(currentState));
        throw; // 抛出异常触发SessionProcessor的错误处理逻辑
    }
}

3. 错误处理与会话清理 #

实现错误处理与会话收尾逻辑，确保处理器稳定运行：

private Task OnProcessErrorAsync(ProcessErrorEventArgs args)
{
    // 记录错误日志，便于排查问题
    _logger.LogError(args.Exception, "Session processing failed: {ErrorSource}", args.ErrorSource);
    return Task.CompletedTask;
}

private async Task OnSessionClosingAsync(SessionClosingEventArgs args)
{
    // 清理会话状态（可选操作）
    await args.SessionReceiver.SetSessionStateAsync(null);
}

// 自定义会话状态类，用于跟踪处理进度与重试次数
public class SessionState
{
    public string Status { get; set; } = "Idle";
    public int RetryCount { get; set; } = 0;
}

关键注意点 #

• 锁时长配置：批量处理耗时较长，必须延长MaxAutoLockRenewalDuration，避免消息锁过期被重新分发
• 事务边界：使用TransactionScope保证业务处理与消息完成操作的原子性，注意这是本地事务，仅作用于当前会话内的操作
• 重试与死信控制：通过会话状态跟踪重试次数，超过阈值后将整个会话的消息移入死信，避免无限重试消耗资源
• 并发限制：通过MaxConcurrentSessions控制同时处理的会话数量，防止服务器资源过载

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