升级MassTransit至5.3.0时出现ScopedConsumeContextProvider未注册错误

问题背景 #

当前使用MassTransit 5.2.0版本，因对接AWS环境RabbitMQ需要amqps协议（5.2.0不支持），计划升级至5.5.6版本，但从5.2.3升级到5.3.0时触发以下错误：

2023-09-14 16:01:09.0917|ERROR|MassTransit.Messages|R-FAULT rabbitmq://localhost/dqportal/dq_audits db270000-1fd6-00ff-3025-08dbb52b0be4
DQ.Audits.IntegrationEvents.Contracts.IAuditEventOccurredIntegrationEvent DQ.Audits.IntegrationEvents.Consumers.AuditEventOccurredConsumer(00:00:00.0013801)
No service for type 'MassTransit.Scoping.ScopedConsumeContextProvider' has been registered.
Exception:System.InvalidOperationException: No service for type 'MassTransit.Scoping.ScopedConsumeContextProvider' has been registered.
at Microsoft.Extensions.DependencyInjection.ServiceProviderServiceExtensions.GetRequiredService(IServiceProvider provider, Type serviceType)
at Microsoft.Extensions.DependencyInjection.ServiceProviderServiceExtensions.GetRequiredService[T](IServiceProvider provider)
at MassTransit.ExtensionsDependencyInjectionIntegration.ScopeProviders.InternalScopeExtensions.UpdateScope(IServiceScope scope, ConsumeContext context)
at MassTransit.ExtensionsDependencyInjectionIntegration.ScopeProviders.DependencyInjectionConsumerScopeProvider.MassTransit.Scoping.IConsumerScopeProvider.GetScope[TConsumer,T](ConsumeContext1 context) at MassTransit.Scoping.ScopeConsumerFactory1.Send[TMessage](ConsumeContext1 context, IPipe1 next)
at MassTransit.Pipeline.Filters.ConsumerMessageFilter2.GreenPipes.IFilter<MassTransit.ConsumeContext<TMessage>>.Send(ConsumeContext1 context, IPipe1 next) at MassTransit.Pipeline.Filters.ConsumerMessageFilter2.GreenPipes.IFilter<MassTransit.ConsumeContext<TMessage>>.Send(ConsumeContext1 context, IPipe1 next)
at GreenPipes.Filters.TeeFilter1.Send(TContext context, IPipe1 next)
at GreenPipes.Filters.OutputPipeFilter2.SendToOutput(TInput context, IPipe1 next, TOutput pipeContext)
at GreenPipes.Filters.OutputPipeFilter2.SendToOutput(TInput context, IPipe1 next, TOutput pipeContext)
at GreenPipes.Filters.DynamicFilter1.Send(TInput context, IPipe1 next)
at MassTransit.Pipeline.Filters.DeserializeFilter.Send(ReceiveContext context, IPipe1 next) at GreenPipes.Filters.RescueFilter2.GreenPipes.IFilter<TContext>.Send(TContext context, IPipe`1 next)

5.2.3为最后正常版本，5.3.0首次出现故障，消费者已简化为仅输出日志，问题不在消费者本身，推测需调整MassTransit注册配置。

原初始化代码 #

public void ConfigureServices(IServiceCollection services)
{
   // other registrations ...
   services.AddEventBus(Configuration);
}

public static class SetupEventBusTask
{
    public static IServiceCollection AddEventBus(this IServiceCollection services, IConfiguration configuration)
    {
        services.Scan(scan => scan.FromAssemblyOf<AuditEventOccurredConsumer>()
            .AddClasses(c => c.Where(t => t.Name.EndsWith("Consumer"))).AsSelf().WithScopedLifetime());

        if (configuration.GetValue<bool>("Queue:Enabled"))
            configureEventBusWithQueue(services, configuration);
        else
            configureEventBusInMemory(services, configuration);

        services.AddSingleton<IBus>(s => s.GetRequiredService<IBusControl>());
        services.AddSingleton<IHostedService, BusService>();
        services.AddScoped<IIntegrationEventStore, IntegrationEventStore>();

        return services;
    }

    private static void configureEventBusWithQueue(IServiceCollection services, IConfiguration configuration)
    {
        services.AddSingleton(provider => Bus.Factory.CreateUsingRabbitMq(cfg =>
        {
            IRabbitMqHost host = cfg.Host(
                configuration.GetValue<string>("Queue:HostName"),
                configuration.GetValue<string>("Queue:VirtualHost"),
                h => {
                    h.Username(configuration.GetValue<string>("Queue:Username"));
                    h.Password(configuration.GetValue<string>("Queue:Password"));
                });

            cfg.ReceiveEndpoint(host, configuration.GetValue<string>("Queue:QueueName"), endpoint =>
            {
                endpoint.Consumer<AuditEventOccurredConsumer>(provider);
            });

            //cfg.UseMessageScheduler(new Uri(configuration.GetValue<string>("Queue:SchedulerQueueUri")));
            cfg.UseNLog();
        }));
    }

    private static void configureEventBusInMemory(IServiceCollection services, IConfiguration configuration)
    {
        services.AddSingleton(provider => Bus.Factory.CreateUsingInMemory(cfg =>
        {
            cfg.UseNLog();
        }));
    }
}

public class BusService : IHostedService
{
    private readonly IBusControl busControl;
    private readonly ILogger<BusService> logger;
    private readonly IConfiguration configuration;

    public BusService(
        IBusControl busControl,
        ILogger<BusService> logger,
        IConfiguration configuration)
    {
        this.busControl = busControl;
        this.logger = logger;
        this.configuration = configuration;
    }

    public async Task StartAsync(CancellationToken cancellationToken)
    {
        logger.LogInformation("Starting eventbus");

        string hostname = configuration.GetValue<string>("Queue:HostName");
        string virtualhost = configuration.GetValue<string>("Queue:VirtualHost");
        string username = configuration.GetValue<string>("Queue:Username");
        string queuename = configuration.GetValue<string>("Queue:QueueName");

        logger.LogInformation($"Configuring Event Bus with Values: {hostname}, {virtualhost}, {username}, {queuename}.");

        await busControl.StartAsync(cancellationToken);

        logger.LogInformation("Started eventbus");
    }

    public async Task StopAsync(CancellationToken cancellationToken)
    {
        logger.LogInformation("Stopping eventbus");
        await busControl.StopAsync(cancellationToken);
        logger.LogInformation("Stopped eventbus");
    }
}

解决方案 #

MassTransit 5.3.0开始，依赖注入集成逻辑调整，需确保ScopedConsumeContextProvider正确注册，推荐使用官方AddMassTransit扩展方法替代手动创建Bus实例，该方法会自动处理所有必要服务的注册。

修改后的代码 #

public static class SetupEventBusTask
{
    public static IServiceCollection AddEventBus(this IServiceCollection services, IConfiguration configuration)
    {
        services.Scan(scan => scan.FromAssemblyOf<AuditEventOccurredConsumer>()
            .AddClasses(c => c.Where(t => t.Name.EndsWith("Consumer"))).AsSelf().WithScopedLifetime());

        services.AddMassTransit(x =>
        {
            // 注册消费者
            x.AddConsumer<AuditEventOccurredConsumer>();

            if (configuration.GetValue<bool>("Queue:Enabled"))
            {
                x.UsingRabbitMq((context, cfg) =>
                {
                    cfg.Host(
                        configuration.GetValue<string>("Queue:HostName"),
                        configuration.GetValue<string>("Queue:VirtualHost"),
                        h => {
                            h.Username(configuration.GetValue<string>("Queue:Username"));
                            h.Password(configuration.GetValue<string>("Queue:Password"));
                        });

                    cfg.ReceiveEndpoint(configuration.GetValue<string>("Queue:QueueName"), endpoint =>
                    {
                        endpoint.ConfigureConsumer<AuditEventOccurredConsumer>(context);
                    });

                    cfg.UseNLog();
                });
            }
            else
            {
                x.UsingInMemory((context, cfg) =>
                {
                    cfg.UseNLog();
                });
            }
        });

        services.AddSingleton<IBus>(s => s.GetRequiredService<IBusControl>());
        services.AddSingleton<IHostedService, BusService>();
        services.AddScoped<IIntegrationEventStore, IntegrationEventStore>();

        return services;
    }
}

关键说明 #

• AddMassTransit是5.3+版本的标准依赖注入方式，会自动注册ScopedConsumeContextProvider等核心服务，避免手动注册遗漏
• 接收端点配置改用ConfigureConsumer方法，这是新版本推荐的消费者配置方式
• 若因特殊情况无法使用AddMassTransit，可手动添加services.AddScoped<ScopedConsumeContextProvider>()来解决注册缺失问题

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