MassTransit：如何避免重复消费同一消息？

针对你遇到的长时间处理任务导致重复消费的问题，核心原因是Azure Service Bus的消息锁过期时间短于任务处理时长，锁过期后消息会重新进入队列被再次消费。以下是具体解决方案：

1. 配置消息锁自动续期与合理时长 #

Azure Service Bus的PeekLock模式下，默认锁时长为60秒，且最大初始锁时长为5分钟。对于5分钟的处理任务，需要启用自动续锁，确保锁在处理期间不会过期。修改你的MassTransit配置：

services.AddMassTransit(busRegistrationConfigurator =>
{
    busRegistrationConfigurator.SetKebabCaseEndpointNameFormatter();

    busRegistrationConfigurator.UsingAzureServiceBus((registrationContext, busFactoryConfigurator) =>
    {
        busFactoryConfigurator.Host(busConnectionString);
        
        // 配置端点时针对长耗时消费者设置锁参数
        busFactoryConfigurator.ConfigureEndpoints(registrationContext, (context, configurator) =>
        {
            // 替换成你的长耗时消费者类型
            if (configurator.ConsumerType == typeof(YourLongRunningConsumer))
            {
                // 初始锁时长设为ASB允许的最大值5分钟
                configurator.LockDuration = TimeSpan.FromMinutes(5);
                // 自动续锁超时设为大于处理时长，比如10分钟，确保处理完成前锁不会过期
                configurator.AutoRenewTimeout = TimeSpan.FromMinutes(10);
            }
        });
    });

    foreach (var implementation in consumerImplementations)
    {
        busRegistrationConfigurator.AddConsumer(implementation);
    }

    busRegistrationConfigurator.AddEntityFrameworkOutbox<TDbContext>(outboxConfigurator =>
    {
        outboxConfigurator.UseSqlServer();
        outboxConfigurator.UseBusOutbox();
    });
});

2. 实现业务幂等性 #

即使消息被重复消费，也要保证业务逻辑不会重复执行。可以通过以下方式实现：

• 在数据库中新增一张消息处理记录表，记录CorrelationId或MessageId的处理状态（已处理/未处理）
• 消费者处理消息前，先查询该记录：
  • 若已处理，直接返回
  • 若未处理，执行业务逻辑，完成后标记为已处理（建议用事务包裹查询与业务操作）

示例代码片段：

public async Task Consume(ConsumeContext<YourMessage> context)
{
    using var transaction = await _dbContext.Database.BeginTransactionAsync();
    try
    {
        // 检查是否已处理
        var existingRecord = await _dbContext.MessageProcessedRecords
            .FirstOrDefaultAsync(x => x.CorrelationId == context.Message.CorrelationId);
        
        if (existingRecord != null)
        {
            await transaction.CommitAsync();
            return;
        }

        // 执行5万条记录的处理逻辑
        ProcessLargeRecords(context.Message.Records);

        // 记录处理状态
        _dbContext.MessageProcessedRecords.Add(new MessageProcessedRecord
        {
            CorrelationId = context.Message.CorrelationId,
            ProcessedAt = DateTime.UtcNow
        });

        await _dbContext.SaveChangesAsync();
        await transaction.CommitAsync();
    }
    catch (Exception)
    {
        await transaction.RollbackAsync();
        throw; // 抛出异常让MassTransit重试（如果需要）
    }
}

3. 拆分大任务为小任务 #

将5万条记录拆分为多个子任务，比如每1000条为一个子任务，发布多个子消息分别处理，最后汇总结果。这样每个子任务处理时间大幅缩短，避免锁过期问题。

示例思路：

• 发布一个"启动大任务"的消息，消费者收到后拆分记录为多个子消息
• 每个子消息消费者处理1000条记录，完成后发布"子任务完成"消息
• 监听"子任务完成"消息，统计所有子任务完成后，标记整个大任务完成

4. 调整Outbox配置（可选） #

你已启用EF Outbox，确保Outbox的清理机制正常，避免重复发送消息。可以调整Outbox的轮询间隔与留存时间：

busRegistrationConfigurator.AddEntityFrameworkOutbox<TDbContext>(outboxConfigurator =>
{
    outboxConfigurator.UseSqlServer();
    outboxConfigurator.UseBusOutbox();
    // 设置轮询间隔，比如10秒
    outboxConfigurator.PollInterval = TimeSpan.FromSeconds(10);
    // 已处理消息留存时间，比如1天
    outboxConfigurator.MessageCleanupInterval = TimeSpan.FromDays(1);
});

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