如何为Rebus消息处理管道添加收发与处理前后中间件并实现指定常规操作？

嘿，刚好在Rebus里折腾过全流程的中间件拦截，这就给你一步步拆解怎么实现你要的功能——从发送前后的用户上下文捕获，到处理前后的验证去重、结果记录，全给你安排明白！

Rebus的消息管道分为发送管道（OutgoingPipeline）和处理管道（IncomingPipeline），我们可以通过自定义中间件步骤，嵌入到管道的不同位置，实现发送/处理各阶段的拦截逻辑。

一、发送管道：消息发送前/后拦截（用户上下文捕获） #

我们需要在消息序列化前，把当前用户的上下文塞进消息头里；发送完成后还可以做日志或监控。

1. 自定义发送中间件 #

// 用户上下文访问器：用来获取当前请求的用户信息
public interface IUserContextAccessor
{
    UserContext CurrentUser { get; set; }
}

public class UserContextAccessor : IUserContextAccessor
{
    public UserContext CurrentUser { get; set; }
}

public class UserContext
{
    public Guid UserId { get; set; }
    public string UserName { get; set; }
}

// 发送前捕获用户上下文的中间件
public class UserContextOutgoingMiddleware : IOutgoingStep
{
    private readonly IUserContextAccessor _userContextAccessor;

    public UserContextOutgoingMiddleware(IUserContextAccessor userContextAccessor)
    {
        _userContextAccessor = userContextAccessor;
    }

    public async Task Process(OutgoingStepContext context, Func<Task> next)
    {
        // 🚀 发送前：捕获当前用户上下文，写入消息头
        var currentUser = _userContextAccessor.CurrentUser;
        if (currentUser != null)
        {
            var messageHeaders = context.Load<Dictionary<string, string>>();
            messageHeaders["X-User-Id"] = currentUser.UserId.ToString();
            messageHeaders["X-User-Name"] = currentUser.UserName;
            // 还可以添加角色、租户ID等更多上下文字段
        }

        // 执行后续发送流程（比如序列化、投递到MQ）
        await next();

        // 🎉 发送后：记录发送成功日志（可选）
        var message = context.Load<Message>();
        Console.WriteLine($"消息 [{message.MessageId}] 已成功发送到MQ");
    }
}

2. 注册发送中间件 #

在Rebus配置时，把中间件插入到序列化步骤之前，确保头信息能被正确序列化：

Configure.With(yourDependencyInjectionActivator)
    .Transport(t => t.UseRabbitMq("amqp://localhost", "your-queue-name"))
    .Options(o =>
    {
        // 注册用户上下文访问器
        o.Register<IUserContextAccessor>(_ => new UserContextAccessor());
        
        // 注入发送中间件
        o.Decorate<IOutgoingPipeline>(pipeline => 
            new PipelineStepInjector(pipeline)
                .OnSend(
                    new UserContextOutgoingMiddleware(yourDependencyInjectionActivator.Get<IUserContextAccessor>()),
                    PipelineRelativePosition.Before,
                    typeof(SerializeOutgoingMessageStep)
                )
        );
    })
    .Start();

二、处理管道：消息处理前/后拦截（上下文恢复+验证去重+结果记录） #

处理管道需要完成三个核心逻辑：恢复用户上下文、验证去重、记录处理结果，中间件的顺序非常关键。

1. 恢复用户上下文中间件 #

在消息反序列化后，从消息头里取出用户上下文，设置到异步本地存储（AsyncLocal）中，避免线程切换导致上下文丢失：

public class UserContextIncomingMiddleware : IIncomingStep
{
    private readonly IUserContextAccessor _userContextAccessor;
    private readonly AsyncLocal<UserContext> _asyncLocalContext = new AsyncLocal<UserContext>();

    public UserContextIncomingMiddleware(IUserContextAccessor userContextAccessor)
    {
        _userContextAccessor = userContextAccessor;
    }

    public async Task Process(IncomingStepContext context, Func<Task> next)
    {
        // 🛠️ 处理前：从消息头恢复用户上下文
        var messageHeaders = context.Load<Dictionary<string, string>>();
        if (messageHeaders.TryGetValue("X-User-Id", out var userIdStr) 
            && Guid.TryParse(userIdStr, out var userId)
            && messageHeaders.TryGetValue("X-User-Name", out var userName))
        {
            var userContext = new UserContext { UserId = userId, UserName = userName };
            _asyncLocalContext.Value = userContext;
            _userContextAccessor.CurrentUser = userContext;
        }

        try
        {
            // 执行后续处理流程（验证、消息处理）
            await next();
        }
        finally
        {
            // 🧹 处理后：清理上下文，防止泄漏到后续请求
            _asyncLocalContext.Value = null;
            _userContextAccessor.CurrentUser = null;
        }
    }
}

2. 消息验证与去重中间件 #

在用户上下文恢复后，先验证消息格式，再通过缓存实现去重（避免重复消费）：

public class MessageValidationAndDeduplicationMiddleware : IIncomingStep
{
    private readonly ICacheService _cacheService; // 自定义缓存服务（内存/Redis都可以）
    private readonly TimeSpan _deduplicationExpiry = TimeSpan.FromHours(24); // 去重缓存过期时间

    public MessageValidationAndDeduplicationMiddleware(ICacheService cacheService)
    {
        _cacheService = cacheService;
    }

    public async Task Process(IncomingStepContext context, Func<Task> next)
    {
        var message = context.Load<Message>();
        var messageId = message.MessageId;

        // ✅ 第一步：消息格式验证
        var messageBody = context.Load<object>();
        if (messageBody is IValidatableMessage validatableMsg)
        {
            var validationResult = validatableMsg.Validate();
            if (!validationResult.IsValid)
            {
                throw new InvalidOperationException($"消息 [{messageId}] 验证失败：{string.Join(", ", validationResult.Errors)}");
            }
        }

        // 🚫 第二步：消息去重
        var cacheKey = $"ProcessedMsg:{messageId}";
        if (await _cacheService.ExistsAsync(cacheKey))
        {
            Console.WriteLine($"消息 [{messageId}] 已处理过，跳过消费");
            return; // 直接跳过后续处理
        }

        try
        {
            await next();
            // 处理成功后，将消息ID存入缓存
            await _cacheService.SetAsync(cacheKey, "processed", _deduplicationExpiry);
        }
        catch (Exception ex)
        {
            Console.WriteLine($"消息 [{messageId}] 处理失败：{ex.Message}");
            throw; // 重新抛出异常，让Rebus处理重试逻辑
        }
    }
}

// 验证消息的接口示例
public interface IValidatableMessage
{
    ValidationResult Validate();
}

public class ValidationResult
{
    public bool IsValid { get; set; }
    public List<string> Errors { get; set; } = new List<string>();
}

3. 处理结果记录中间件 #

在消息处理完成后（无论成功失败），记录处理时长、结果等信息：

public class ProcessingResultLoggingMiddleware : IIncomingStep
{
    private readonly ILogger<ProcessingResultLoggingMiddleware> _logger;

    public ProcessingResultLoggingMiddleware(ILogger<ProcessingResultLoggingMiddleware> logger)
    {
        _logger = logger;
    }

    public async Task Process(IncomingStepContext context, Func<Task> next)
    {
        var message = context.Load<Message>();
        var messageId = message.MessageId;
        var startTime = DateTime.UtcNow;
        bool isSuccess = true;
        Exception exception = null;

        try
        {
            await next();
        }
        catch (Exception ex)
        {
            isSuccess = false;
            exception = ex;
            throw;
        }
        finally
        {
            var duration = DateTime.UtcNow - startTime;
            // 📝 记录处理结果到日志/数据库
            if (isSuccess)
            {
                _logger.LogInformation("消息 [{MessageId}] 处理成功，耗时 {DurationMs}ms", messageId, duration.TotalMilliseconds);
            }
            else
            {
                _logger.LogError(exception, "消息 [{MessageId}] 处理失败，耗时 {DurationMs}ms", messageId, duration.TotalMilliseconds);
            }
        }
    }
}

4. 注册处理管道中间件 #

按照上下文恢复 → 验证去重 → 结果记录的顺序注册，确保逻辑执行顺序正确：

Configure.With(yourDependencyInjectionActivator)
    .Transport(t => t.UseRabbitMq("amqp://localhost", "your-queue-name"))
    .Options(o =>
    {
        // 注册依赖服务
        o.Register<IUserContextAccessor>(_ => new UserContextAccessor());
        o.Register<ICacheService>(_ => new InMemoryCacheService()); // 替换为你的缓存实现
        
        // 注入处理管道中间件
        o.Decorate<IIncomingPipeline>(pipeline => 
            new PipelineStepInjector(pipeline)
                // 1. 反序列化后恢复用户上下文
                .OnReceive(
                    new UserContextIncomingMiddleware(yourDependencyInjectionActivator.Get<IUserContextAccessor>()),
                    PipelineRelativePosition.After,
                    typeof(DeserializeIncomingMessageStep)
                )
                // 2. 验证去重
                .OnReceive(
                    new MessageValidationAndDeduplicationMiddleware(yourDependencyInjectionActivator.Get<ICacheService>()),
                    PipelineRelativePosition.After,
                    typeof(UserContextIncomingMiddleware)
                )
                // 3. 处理完成后记录结果
                .OnReceive(
                    new ProcessingResultLoggingMiddleware(yourDependencyInjectionActivator.Get<ILogger<ProcessingResultLoggingMiddleware>>()),
                    PipelineRelativePosition.After,
                    typeof(HandleMessageStep)
                )
        );
    })
    .Start();

关键注意事项 #

• 中间件顺序：一定要按照逻辑依赖顺序注册，比如必须先恢复用户上下文，才能在验证时使用用户信息；
• 异步上下文存储：必须用AsyncLocal而非ThreadStatic，Rebus的消息处理是异步的，线程会频繁切换；
• 缓存过期：去重缓存要设置合理的过期时间，避免内存占用过高；
• 上下文清理：用户上下文必须在finally块中清理，防止泄漏到后续请求。

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