使用MassTransit多总线时如何共享Scope上下文？

针对你在多总线场景下无法通过过滤器共享CorrelationId的问题，提供以下几种可行方案：

方案1：利用MassTransit内置ConsumeContext传递（推荐） #

直接复用MassTransit消费者上下文自带的CorrelationId，无需自定义Scoped上下文，避免多总线的DI范围隔离问题：

修改过滤器实现 #

public class CorrelationSendFilter<T> : IFilter<SendContext<T>>
    where T : class
{
    private readonly ConsumeContext _consumeContext;

    public CorrelationSendFilter(ConsumeContext consumeContext) 
    { 
        _consumeContext = consumeContext; 
    }

    public async Task Send(SendContext<T> context, IPipe<SendContext<T>> next)
    {
        // 复用当前消费的CorrelationId，无则生成新ID
        context.CorrelationId = _consumeContext.CorrelationId ?? Guid.NewGuid();
        await next.Send(context);
    }

    public void Probe(ProbeContext context) { }
}

多总线配置时注册过滤器 #

为每个总线的发送管道添加该过滤器：

// 配置第一个总线
services.AddMassTransit(busConfig =>
{
    busConfig.AddConsumer<YourFirstConsumer>();

    busConfig.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("rabbitmq://localhost");
        cfg.ConfigureEndpoints(context);

        // 全局注册发送过滤器
        cfg.SendTopology.UseFilter(typeof(CorrelationSendFilter<>));
    });
});

// 配置第二个总线
services.AddMassTransit("SecondBus", busConfig =>
{
    busConfig.AddConsumer<YourSecondConsumer>();

    busConfig.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("rabbitmq://localhost");
        cfg.ConfigureEndpoints(context);

        // 同样注册发送过滤器
        cfg.SendTopology.UseFilter(typeof(CorrelationSendFilter<>));
    });
});

方案2：用AsyncLocal改造自定义CorrelationContext #

如果必须保留自定义的CorrelationContext，将其内部存储改为AsyncLocal，实现异步流内的CorrelationId共享：

改造CorrelationContext #

public class CorrelationContext
{
    private readonly AsyncLocal<Guid?> _asyncLocalCorrelationId = new AsyncLocal<Guid?>();

    public Guid GetOrNewCorrelationId()
    {
        if (_asyncLocalCorrelationId.Value == null)
            _asyncLocalCorrelationId.Value = Guid.NewGuid();
        return _asyncLocalCorrelationId.Value.Value;
    }

    // 从消费上下文初始化CorrelationId
    public void InitFromConsumeContext(ConsumeContext consumeContext)
    {
        if (consumeContext.CorrelationId.HasValue)
            _asyncLocalCorrelationId.Value = consumeContext.CorrelationId.Value;
    }
}

消费者中初始化CorrelationId #

public class YourConsumer : IConsumer<YourMessage>
{
    private readonly CorrelationContext _correlationContext;

    public YourConsumer(CorrelationContext correlationContext)
    {
        _correlationContext = correlationContext;
    }

    public async Task Consume(ConsumeContext<YourMessage> context)
    {
        // 从消费上下文初始化CorrelationId
        _correlationContext.InitFromConsumeContext(context);

        // 处理消息并发送新消息，过滤器会自动获取当前异步流的CorrelationId
        await context.Send<AnotherMessage>(new { /* 消息内容 */ });
    }
}

注册服务与过滤器 #

// 注册CorrelationContext为Scoped或Singleton（AsyncLocal天然隔离异步流）
services.AddScoped<CorrelationContext>();

// 每个总线配置时注册过滤器
cfg.SendTopology.UseFilter(typeof(CorrelationSendFilter<>));

方案3：为每个总线单独绑定过滤器DI范围 #

在多总线配置过程中，直接从当前总线的DI上下文解析过滤器，确保Scoped服务正确注入：

// 第一个总线配置
services.AddMassTransit(busConfig =>
{
    busConfig.AddConsumer<YourFirstConsumer>();

    busConfig.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("rabbitmq://localhost");
        cfg.ConfigureEndpoints(context);

        // 针对特定消息类型注册过滤器
        cfg.Send<YourMessage>(x => x.UseFilter(context.GetRequiredService<CorrelationSendFilter<YourMessage>>()));
    });
});

// 第二个总线配置
services.AddMassTransit("SecondBus", busConfig =>
{
    busConfig.AddConsumer<YourSecondConsumer>();

    busConfig.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("rabbitmq://localhost");
        cfg.ConfigureEndpoints(context);

        cfg.Send<AnotherMessage>(x => x.UseFilter(context.GetRequiredService<CorrelationSendFilter<AnotherMessage>>()));
    });
});

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