MassTransit配置订阅端点时如何获取IRegistrationContext注册消费者

问题原因 #

你没遗漏核心API，只是没拿到DI注册上下文：

• 直接从总线解析的ISubscriptionEndpointConnector属于基础连接组件，创建动态端点时不会自动携带DI注册上下文，所以没法直接调用无参的ConfigureConsumer<TConsumer>()扩展方法——这个无参方法要求配置器自身持有IRegistrationContext才能完成消费者的依赖注入配置。
• 你之前用的IReceiveEndpointConnector是DI注册阶段封装过的版本，自带注册上下文，所以配置委托里会直接暴露IBusRegistrationContext参数，不需要手动传。

解决方法 #

不需要找其他特殊API，直接从DI容器解析IBusRegistrationContext就行，这个对象在你调用AddMassTransit注册服务的时候就已经自动加入DI容器了，不需要额外配置。

1. 在负责管理消费者启停的服务构造函数里，同时注入ISubscriptionEndpointConnector和IBusRegistrationContext
2. 配置订阅端点时，把注入拿到的IBusRegistrationContext作为参数传入ConfigureConsumer方法

可直接运行的代码示例 #

public class ConsumerSwitchService
{
    private readonly ISubscriptionEndpointConnector _subscriptionConnector;
    private readonly IBusRegistrationContext _registrationContext;
    private HostReceiveEndpointHandle? _runningConsumerHandle;

    public ConsumerSwitchService(
        ISubscriptionEndpointConnector subscriptionConnector,
        IBusRegistrationContext registrationContext)
    {
        _subscriptionConnector = subscriptionConnector;
        _registrationContext = registrationContext;
    }

    // 业务条件满足时启动消费者
    public async Task StartFilteredConsumerAsync(CancellationToken ct)
    {
        if (_runningConsumerHandle != null) return;

        _runningConsumerHandle = _subscriptionConnector.ConnectSubscriptionEndpoint<SomeMessage>(
            "business-subscription",
            cfg =>
            {
                // 配置SQL过滤规则
                cfg.Rule = new CreateRuleOptions
                {
                    Name = "OnlySpecialState",
                    Filter = new SqlRuleFilter("state = 'Special'")
                };
                // 传入注册上下文完成消费者配置，支持所有DI依赖注入
                cfg.ConfigureConsumer<SomeConsumer>(_registrationContext);
                // 其他常规配置和普通接收端点完全一致
                cfg.MaxConcurrentCalls = 10;
                cfg.PrefetchCount = 20;
            });

        // 等待端点就绪再返回
        await _runningConsumerHandle.Ready.WaitAsync(ct);
    }

    // 业务条件不满足时停止消费者
    public async Task StopFilteredConsumerAsync(CancellationToken ct)
    {
        if (_runningConsumerHandle == null) return;

        await _runningConsumerHandle.StopAsync(ct);
        _runningConsumerHandle = null;
    }
}

补充说明 #

• 这种方式配置的消费者和静态启动的消费者没有行为差异，消费者构造函数里注入的DbContext、业务服务、配置类等都能正常解析，重试策略、死信配置、并发控制等都可以在配置委托里正常设置。
• 如果需要配置多个过滤规则，不要用cfg.Rule单属性赋值（会覆盖已有规则），通过cfg.Rules.Add()方法追加多个CreateRuleOptions即可。
• 启停返回的句柄是HostReceiveEndpointHandle类型，和IReceiveEndpointConnector返回的句柄用法完全一致，停止时会等待当前正在处理的消息执行完成再断开连接，不会造成消息中断丢失。

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