MassTransit 7及以上版本中CreateBrokeredMessageReceiver的替代方案是什么？

问题背景 #

我正将.NET 3.1应用升级到.NET 6，该应用基于CQRS模式，用Azure Service Bus和函数应用处理命令并分发给对应消费者。之前在.NET 3.1里用Bus.Factory.CreateBrokeredMessageReceiver构建处理程序，但升级到.NET 6并把所有MassTransit包更新到v7后，这个方法被移除了。我有大量消费者，单消费者方案不适用。已经升级了MassTransit.Azure.ServiceBus.Core、MassTransit.Extensions.DependencyInjection、MassTransit.WebJobs.ServiceBus这些包，也试过网上提到的IMessageReceiver替代方案但没成功，求可行的替代方法。

原代码示例：

[FunctionName("MessageHandler")]
public async Task Run(
    [ServiceBusTrigger("%ServiceBusQueue%", Connection = "ServiceBusConnection")]
        ServiceBusReceivedMessage message,
        CancellationToken cancellationToken)
{
    ...
    var handler = Bus.Factory.CreateBrokeredMessageReceiver(
        binder,
        cfg =>
        {
            cfg.CancellationToken = cancellationToken;
            cfg.SetLog(logger);
            cfg.InputAddress = new Uri(_settings.ServiceBusUrl);
            cfg.UseRetry(x => x.Intervals(2000, 4000, 8000, 16000));
            cfg.Consumer(GetConsumerType(messageType), type => _serviceProvider.GetService(type));
        });

    await handler.Handle(message);
    ...
}

private Type GetConsumerType(string messageType)
{
    ...
    return consumerType; 
}

替代方案：使用ServiceBusMessageReceiver和IConsumerFactory #

在MassTransit v7+中，CreateBrokeredMessageReceiver被官方替换为ServiceBusMessageReceiver，结合依赖注入的DependencyInjectionConsumerFactory可以动态解析大量消费者，完美适配原有场景。

1. 调整函数处理逻辑 #

修改原函数代码，用新的接收器和工厂替代旧实现，保持原有重试策略和动态消费者解析逻辑：

using MassTransit;
using MassTransit.Azure.ServiceBus.Core;
using MassTransit.Context;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.ServiceBus;

[FunctionName("MessageHandler")]
public async Task Run(
    [ServiceBusTrigger("%ServiceBusQueue%", Connection = "ServiceBusConnection")]
        ServiceBusReceivedMessage message,
    CancellationToken cancellationToken)
{
    // 保留原有消息类型解析逻辑
    var messageType = ...; // 你的现有逻辑
    var consumerType = GetConsumerType(messageType);

    // 从DI容器创建消费者工厂
    var consumerFactory = new DependencyInjectionConsumerFactory(_serviceProvider);

    // 初始化ServiceBus消息接收器
    var receiver = new ServiceBusMessageReceiver(
        _serviceProvider.GetRequiredService<ILoggerFactory>(),
        consumerFactory,
        new ServiceBusMessageReceiverOptions
        {
            CancellationToken = cancellationToken,
            InputAddress = new Uri(_settings.ServiceBusUrl),
        });

    // 配置重试策略，和原有逻辑一致
    var retryPolicy = Retry.CreatePolicy(x => x.Intervals(2000, 4000, 8000, 16000));

    // 创建消费上下文并执行处理
    var context = new ServiceBusReceivedMessageContext(message, cancellationToken);
    await using var consumeContextScope = new ConsumeContextScope(context);

    await retryPolicy.ExecuteAsync(async () =>
    {
        await receiver.Consume(consumeContextScope, consumerType);
    });
}

private Type GetConsumerType(string messageType)
{
    // 保留原有消费者类型匹配逻辑
    ...
    return consumerType; 
}

2. 确保DI容器正确注册所有消费者 #

在.NET 6的Program.cs中，通过MassTransit的DI扩展批量注册所有消费者，保证DependencyInjectionConsumerFactory能正确解析实例：

var builder = Host.CreateDefaultBuilder(args);
builder.ConfigureServices(services =>
{
    // 注册MassTransit及Azure Service Bus
    services.AddMassTransit(x =>
    {
        // 批量注册当前程序集内的所有消费者
        x.AddConsumers(typeof(Program).Assembly);

        x.UsingAzureServiceBus((context, cfg) =>
        {
            cfg.Host(Environment.GetEnvironmentVariable("ServiceBusConnection"));
            // 其他Service Bus配置（如队列、主题绑定等）
        });
    });

    // 注册其他业务服务...
});

var host = builder.Build();
host.Run();

关键说明 #

• ServiceBusMessageReceiver是MassTransit v7+针对Azure Service Bus新SDK（ServiceBusReceivedMessage）推出的官方接收器实现，完全替代旧的BrokeredMessageReceiver。
• DependencyInjectionConsumerFactory直接对接DI容器，支持动态解析任意消费者类型，无需为每个消费者单独配置，适配大量消费者场景。
• 重试策略通过Retry.CreatePolicy单独配置，和原有逻辑行为保持一致。

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