基于MassTransit的面向对象出站消息Header注入方案咨询

需要为每条出站消息添加包含AWS环境信息的Header，具体需添加EC2实例ID和AWS账户ID。此前使用MassTransit配置lambda实现Header注入的方式如下：

var busControl = Bus.Factory.CreateUsingInMemory(c =>
  c.ConfigurePublish(x => x.UseExecute(ctx =>
  {
    ctx.Headers.Set("awsAccountId", "abc");
  });

但该逻辑耦合在组合根中，且针对不同发布者（如内存、SQS）需要重复编写lambda代码，希望找到更优的面向对象实现方案，适配DI容器注册，避免重复逻辑。

1. 封装Header注入逻辑为自定义中间件 #

通过实现MassTransit的过滤器接口，把AWS环境Header的注入逻辑封装成独立组件，解耦配置代码：

public class AwsEnvironmentHeaderFilter : IFilter<PublishContext>
{
    private readonly IAwsEnvironmentInfoProvider _awsInfoProvider;

    // 通过DI注入环境信息获取服务，解耦具体实现
    public AwsEnvironmentHeaderFilter(IAwsEnvironmentInfoProvider awsInfoProvider)
    {
        _awsInfoProvider = awsInfoProvider;
    }

    public async Task Send(PublishContext context, IPipe<PublishContext> next)
    {
        var awsInfo = await _awsInfoProvider.GetEnvironmentInfoAsync();
        
        // 注入Header
        context.Headers.Set("awsAccountId", awsInfo.AccountId);
        context.Headers.Set("ec2InstanceId", awsInfo.InstanceId);

        // 执行后续管道逻辑
        await next.Send(context);
    }

    public void Probe(ProbeContext context)
    {
        context.CreateFilterScope("aws-environment-header");
    }
}

// 定义环境信息获取接口，便于测试和扩展
public interface IAwsEnvironmentInfoProvider
{
    Task<AwsEnvironmentInfo> GetEnvironmentInfoAsync();
}

public class AwsEnvironmentInfo
{
    public string AccountId { get; set; }
    public string InstanceId { get; set; }
}

// 实际AWS环境信息获取实现
public class AwsEnvironmentInfoProvider : IAwsEnvironmentInfoProvider
{
    public async Task<AwsEnvironmentInfo> GetEnvironmentInfoAsync()
    {
        return new AwsEnvironmentInfo
        {
            AccountId = await FetchAwsAccountId(),
            InstanceId = await FetchEc2InstanceId()
        };
    }

    private async Task<string> FetchAwsAccountId()
    {
        // 通过STS服务获取当前账户ID
        using var stsClient = new AmazonSecurityTokenServiceClient();
        var response = await stsClient.GetCallerIdentityAsync();
        return response.Account;
    }

    private async Task<string> FetchEc2InstanceId()
    {
        // 从EC2实例元数据服务获取实例ID
        using var client = new HttpClient();
        client.DefaultRequestHeaders.Add("X-aws-ec2-metadata-token", await FetchMetadataToken());
        var instanceId = await client.GetStringAsync("http://169.254.169.254/latest/meta-data/instance-id");
        return instanceId.Trim();
    }

    private async Task<string> FetchMetadataToken()
    {
        using var client = new HttpClient();
        var request = new HttpRequestMessage(HttpMethod.Put, "http://169.254.169.254/latest/api/token");
        request.Headers.Add("X-aws-ec2-metadata-token-ttl-seconds", "3600");
        var response = await client.SendAsync(request);
        return await response.Content.ReadAsStringAsync();
    }
}

2. 编写通用扩展方法简化注册 #

创建MassTransit扩展方法，让不同传输方式（内存、SQS）可以复用Header注入逻辑：

public static class MassTransitAwsHeaderExtensions
{
    public static void AddAwsEnvironmentHeaders(this IPublishPipeConfigurator configurator)
    {
        configurator.AddFilter(new AwsEnvironmentHeaderSpecification());
    }

    private class AwsEnvironmentHeaderSpecification : IPublishPipeSpecification
    {
        public void Apply(IPipeBuilder<PublishContext> builder)
        {
            // 从DI容器中获取环境信息提供者实例
            var provider = builder.GetDependency<IAwsEnvironmentInfoProvider>();
            builder.AddFilter(new AwsEnvironmentHeaderFilter(provider));
        }

        public void Probe(ProbeContext context)
        {
            context.CreateScope("aws-environment-headers");
        }
    }
}

3. 注册服务到DI容器 #

在DI容器中注册环境信息提供者，确保中间件能获取到依赖：

// Program.cs 或 Startup 类中
services.AddScoped<IAwsEnvironmentInfoProvider, AwsEnvironmentInfoProvider>();

4. 配置MassTransit使用自定义中间件 #

无论使用内存总线还是SQS总线，只需调用扩展方法即可启用Header注入，无需重复编写逻辑：

内存总线配置示例 #

services.AddMassTransit(x =>
{
    // 注册消费者等业务逻辑...

    x.UsingInMemory((context, cfg) =>
    {
        // 启用AWS环境Header注入
        cfg.ConfigurePublish(p => p.AddAwsEnvironmentHeaders());
        
        cfg.ConfigureEndpoints(context);
    });
});

SQS总线配置示例 #

services.AddMassTransit(x =>
{
    // 注册消费者等业务逻辑...

    x.UsingAmazonSqs((context, cfg) =>
    {
        cfg.Host("us-east-1");
        
        // 同样启用Header注入
        cfg.ConfigurePublish(p => p.AddAwsEnvironmentHeaders());
        
        cfg.ConfigureEndpoints(context);
    });
});

方案优势 #

• 解耦逻辑：Header注入逻辑独立封装，与组合根分离，符合单一职责原则
• 复用性强：通过扩展方法，所有MassTransit传输方式都能一键启用该功能，避免重复代码
• 适配DI：依赖注入环境信息提供者，便于单元测试（可注入模拟实现）
• 易于维护：后续修改Header规则只需调整AwsEnvironmentHeaderFilter，无需改动多处配置

内容的提问来源于stack exchange，提问作者void.pointer