如何在MassTransit中实现带重试判断的MongoDb自定义持久化收件箱？

核心思路 #

放弃使用IRetryObserver，转而在你已经实现的作用域过滤器中间件内完成重试状态判断和MongoDb收件箱更新——因为过滤器本身是作用域的，天然支持注入作用域仓库，同时能直接获取重试上下文信息。

具体实现步骤 #

1. 在作用域过滤器中获取重试上下文并判断是否终止重试 #

在过滤器的Send方法中，通过ConsumeContext的扩展方法GetRetryContext()获取重试配置和当前重试次数，直接判断是否已达重试上限：

public class InboxFilter : IFilter<ConsumeContext>
{
    private readonly IInboxRepository _inboxRepository;

    // 注入作用域的MongoDb收件箱仓库
    public InboxFilter(IInboxRepository inboxRepository)
    {
        _inboxRepository = inboxRepository;
    }

    public async Task Send(ConsumeContext context, IPipe<ConsumeContext> next)
    {
        try
        {
            await next.Send(context);
        }
        catch (Exception)
        {
            // 获取重试上下文（仅当启用重试策略时存在）
            var retryContext = context.GetRetryContext();
            if (retryContext != null)
            {
                // 判断是否是最后一次重试：当前重试次数+1 >= 配置的重试上限
                var isFinalRetry = retryContext.RetryAttempt + 1 >= retryContext.RetryLimit;
                if (isFinalRetry)
                {
                    // 将MongoDb收件箱中的对应消息标记为dead
                    await _inboxRepository.MarkMessageAsDead(context.MessageId, retryContext.RetryAttempt);
                }
            }
            throw; // 继续抛出异常触发延迟重投递
        }
    }

    public void Probe(ProbeContext context) => context.CreateFilterScope("inbox-filter");
}

2. 正确注册过滤器与重试策略 #

确保过滤器被注册为作用域服务，并在消费者配置中添加到管道，同时配置延迟重投递：

// 注册作用域过滤器和MongoDb仓库
services.AddScoped<InboxFilter>();
services.AddScoped<IInboxRepository, MongoDbInboxRepository>();

// MassTransit消费者配置
cfg.Consumer<YourEventConsumer>(c =>
{
    // 添加自定义作用域过滤器
    c.UseFilter(new InboxFilterSpecification());
    // 配置延迟重投递重试策略，指定重试上限
    c.UseDelayedRedelivery(r => r
        .Intervals(TimeSpan.FromSeconds(10), TimeSpan.FromSeconds(30), TimeSpan.FromMinutes(1))
        .RetryLimit(3)); // 这里的3就是你需要的重试上限
});

3. 收件箱Worker的处理逻辑优化 #

Worker扫描待处理消息时，增加判断逻辑：

• 如果消息未标记为dead，且当前记录的重试次数小于配置的重试上限，则继续发布消息
• 如果消息已标记为dead或重试次数已达上限，则跳过处理，触发告警（比如推送监控指标、发送告警通知）

关键细节说明 #

• RetryAttempt是从0开始计数的：比如配置RetryLimit=3，重试过程中RetryAttempt会依次为0、1、2，第三次尝试失败后触发isFinalRetry判断
• 若GetRetryContext()返回null，说明当前是首次消费，还未进入重试流程，无需标记dead
• 延迟重投递每次都会经过过滤器，所以能准确捕获每次重试的状态，即使应用宕机重启，Worker也能通过MongoDb中的dead标记和重试次数判断是否继续处理

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