基于MassTransit消息内容标注Application Insights遥测以跟踪账户配额请求

核心错误点 #

1. 作用域ITelemetryInitializer的误用 #

ITelemetryInitializer的设计初衷就是单例生命周期，遥测系统在启动阶段就会初始化所有注册的初始化器，后续不会随DI作用域重新创建实例。你将其注册为作用域/瞬时类型，导致初始化器无法关联到当前消息处理的作用域，自然无法获取作用域内的跟踪器数据，这是作用域功能异常的根本原因。

2. TelemetryClient操作的错误使用 #

• 自定义OperationTelemetry子类并非官方推荐的扩展方式，遥测系统的操作生命周期与创建它的线程/上下文绑定，跨线程调用StopOperation必然引发错误。
• RequestTelemetry是专门用于跟踪HTTP请求的遥测类型，不适用于你场景中的出站消息请求，属于类型误用。

3. ITelemetryProcessorFactory的方向正确，但需注意实现细节 #

该工厂确实可以创建能访问作用域服务的ITelemetryProcessor，但关键是要让处理器能获取到当前MassTransit消息处理的DI作用域，而不是简单注入单例服务。

正确实现方案 #

步骤1：定义作用域账户上下文服务 #

先创建一个作用域服务，用于在消息处理周期中存储当前账户信息：

public interface IAccountContext
{
    string AccountId { get; set; }
}

public class AccountContext : IAccountContext
{
    public string AccountId { get; set; }
}

步骤2：在MassTransit消费者中注入并设置账户信息 #

在消息消费时，从消息中提取账户信息并写入作用域上下文：

public class MyMessageConsumer : IConsumer<QuotaLimitedMessage>
{
    private readonly IAccountContext _accountContext;

    public MyMessageConsumer(IAccountContext accountContext)
    {
        _accountContext = accountContext;
    }

    public async Task Consume(ConsumeContext<QuotaLimitedMessage> context)
    {
        // 从消息体中提取账户ID
        _accountContext.AccountId = context.Message.AccountId;

        // 执行出站请求逻辑...
    }
}

步骤3：实现自定义ITelemetryProcessor #

在处理器中从当前作用域获取账户信息，添加到遥测属性：

public class AccountQuotaTelemetryProcessor : ITelemetryProcessor
{
    private readonly ITelemetryProcessor _nextProcessor;
    private readonly IServiceProvider _serviceProvider;

    public AccountQuotaTelemetryProcessor(ITelemetryProcessor nextProcessor, IServiceProvider serviceProvider)
    {
        _nextProcessor = nextProcessor;
        _serviceProvider = serviceProvider;
    }

    public void Process(ITelemetry item)
    {
        // 从当前DI作用域中获取账户上下文
        if (_serviceProvider.GetService<IAccountContext>() is { } accountContext && !string.IsNullOrEmpty(accountContext.AccountId))
        {
            // 给遥测项添加账户ID属性
            item.Properties["QuotaAccountId"] = accountContext.AccountId;
            
            // 如果是出站请求遥测，可同步到上下文属性
            if (item is RequestTelemetry requestTelemetry)
            {
                requestTelemetry.Context.Properties["QuotaAccountId"] = accountContext.AccountId;
            }
        }

        // 传递给下一个处理器
        _nextProcessor.Process(item);
    }
}

步骤4：实现ITelemetryProcessorFactory并注册 #

通过工厂创建处理器，确保能注入IServiceProvider以访问作用域服务：

public class AccountTelemetryProcessorFactory : ITelemetryProcessorFactory
{
    private readonly IServiceProvider _serviceProvider;

    public AccountTelemetryProcessorFactory(IServiceProvider serviceProvider)
    {
        _serviceProvider = serviceProvider;
    }

    public ITelemetryProcessor Create(ITelemetryProcessor next)
    {
        return new AccountQuotaTelemetryProcessor(next, _serviceProvider);
    }
}

步骤5：DI与遥测配置 #

在Autofac和遥测配置中完成注册：

// Autofac模块注册作用域服务与消费者
public class QuotaTrackingModule : Module
{
    protected override void Load(ContainerBuilder builder)
    {
        // 注册账户上下文为作用域生命周期
        builder.RegisterType<AccountContext>().As<IAccountContext>().InstancePerLifetimeScope();
        
        // 注册MassTransit消费者
        builder.RegisterType<MyMessageConsumer>().AsConsumer();
    }
}

// OpenTelemetry配置中添加处理器工厂
services.AddOpenTelemetry()
    .WithTracing(tracingBuilder => tracingBuilder
        .AddProcessor(sp => new AccountQuotaTelemetryProcessor(sp.GetRequiredService<ITelemetryProcessor>(), sp))
        // 其他遥测配置（如导出器、数据源）...
    );

方案原理 #

MassTransit的每个消息消费都会创建独立的DI作用域，AccountContext会随作用域生命周期存在。AccountQuotaTelemetryProcessor在处理遥测项时，通过IServiceProvider获取当前作用域的AccountContext，从而将账户信息绑定到对应的出站请求遥测中，完美匹配你的配额跟踪需求。

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