使用Testcontainers集成测试时如何启动MassTransit总线？

问题场景 #

使用MassTransit+RabbitMQ（Docker容器）作为消息层的API，基于WebApplicationFactory和Testcontainers编写集成测试（不使用MassTransit内存模拟器），遇到以下问题：

• 调用IRabbitMqBusFactoryConfigurator.ConfigureEndpoints时，端点/队列/交换机仅完成配置映射，未主动声明、绑定，总线也未启动（正常启动API时会一次性完成所有队列和消费者绑定）。
• 仅在调用ISendEndpointProvider.Send时，才会声明对应端点的队列/交换机，但此时队列中有消息，消费者也不会触发执行。

调用Send时的日志：

核心原因分析 #

1. WebApplicationFactory未自动完成总线启动流程：MassTransit总线以IHostedService形式注册，测试场景中主机初始化时机可能导致总线未完成端点声明、消费者绑定的全流程。
2. 配置读取的生命周期冲突：在MassTransit配置回调中创建scope获取IOptionsSnapshot，会导致配置读取时机与总线初始化逻辑不匹配，影响队列自动声明逻辑。
3. 未等待总线就绪：测试中未等待总线完成所有端点初始化就发送消息，此时消费者尚未就绪，无法处理队列中的消息。

解决方案 #

1. 显式启动总线并等待就绪 #

在测试初始化阶段，手动获取IBusControl实例并启动，同时等待总线完成所有端点的声明与绑定：

public class YourIntegrationTests : IAsyncLifetime
{
    private readonly WebApplicationFactory<Program> _factory;
    private readonly RabbitMqContainer _rabbitMqContainer;
    private IBusControl _bus;

    public YourIntegrationTests()
    {
        // 初始化Testcontainers RabbitMQ容器
        _rabbitMqContainer = new RabbitMqBuilder()
            .WithImage("rabbitmq:3-management")
            .Build();

        // 初始化WebApplicationFactory，覆盖配置指向测试容器
        _factory = new WebApplicationFactory<Program>()
            .WithWebHostBuilder(builder =>
            {
                builder.ConfigureAppConfiguration((context, config) =>
                {
                    config.AddInMemoryCollection(new Dictionary<string, string>
                    {
                        ["QueueSettings:Hostname"] = _rabbitMqContainer.GetConnectionString(),
                        ["ServiceSettings:ServiceName"] = "test-service"
                    });
                });
            });
    }

    public async Task InitializeAsync()
    {
        await _rabbitMqContainer.StartAsync();
        // 获取总线实例并启动
        _bus = _factory.Services.GetRequiredService<IBusControl>();
        await _bus.StartAsync(default);
        // 等待总线完成所有端点初始化、消费者绑定
        await _bus.WaitForReadyAsync(default);
    }

    public async Task DisposeAsync()
    {
        await _bus.StopAsync(default);
        await _rabbitMqContainer.StopAsync();
        await _factory.DisposeAsync();
    }
}

2. 修正MassTransit配置的读取逻辑 #

将配置中使用的IOptionsSnapshot替换为IOptions<QueueSettings>，避免范围性服务在单例总线配置中的生命周期冲突：

private static void ConfigureMassTransitWithRabbitMq<TMarker>(this IServiceCollection services, IConfiguration configuration)
    where TMarker : IMarker
{
    services.Configure<QueueSettings>(configuration.GetSection(QueueSettings.SettingName));

    services.AddSingleton<IEndpointNameFormatter>(
        static serviceProvider =>
        {
            var serviceSettings = serviceProvider.GetRequiredService<IOptions<ServiceSettings>>();
            return new KebabCaseEndpointNameFormatter(serviceSettings.Value.ServiceName, false);
        });

    services.AddMassTransit(
        static configure =>
        {
            configure.AddConsumers(Assembly.GetExecutingAssembly(), typeof(TMarker).Assembly);

            configure.UsingRabbitMq(
                static (context, configurator) =>
                {
                    // 直接从根服务提供者获取IOptions，无需创建scope
                    var serviceSettings = context.GetRequiredService<IOptions<ServiceSettings>>().Value;
                    var queueSettings = context.GetRequiredService<IOptions<QueueSettings>>().Value;

                    configurator.Host(queueSettings.Hostname);

                    configurator.UseInstrumentation(serviceName: serviceSettings.ServiceName);

                    configurator.UseMessageRetry(
                        retryConfigurator =>
                        {
                            retryConfigurator.Interval(
                                queueSettings.Retries ?? throw new ArgumentNullException(nameof(queueSettings.Retries)),
                                queueSettings.RetriesTimeSeconds ?? throw new ArgumentNullException(nameof(queueSettings.RetriesTimeSeconds)));
                        });

                    configurator.ClearSerialization();
                    configurator.UseRawJsonSerializer();
                    configurator.ConfigureJsonSerializerOptions(
                        static options =>
                        {
                            var serializerOptions = Application.UI.Dependencies.JsonSerializerOptions;

                            options.IncludeFields = serializerOptions.IncludeFields;
                            options.PropertyNameCaseInsensitive = serializerOptions.PropertyNameCaseInsensitive;

                            foreach (var converter in serializerOptions.Converters)
                            {
                                options.Converters.Add(converter);
                            }

                            options.TypeInfoResolver = serializerOptions.TypeInfoResolver;

                            return options;
                        });

                    configurator.ConfigureEndpoints(context, context.GetRequiredService<IEndpointNameFormatter>());
                });
        });
}

3. 验证消费者消息接收 #

通过注入自定义跟踪服务，在测试中验证消费者是否正确处理消息：

// 定义测试用跟踪服务
public interface ITestMessageTracker
{
    event EventHandler MessageReceived;
    void OnMessageReceived();
}

public class TestMessageTracker : ITestMessageTracker
{
    public event EventHandler MessageReceived;

    public void OnMessageReceived()
    {
        MessageReceived?.Invoke(this, EventArgs.Empty);
    }
}

// 在测试的ConfigureServices中注册跟踪服务
_factory = new WebApplicationFactory<Program>()
    .WithWebHostBuilder(builder =>
    {
        builder.ConfigureServices(services =>
        {
            services.AddSingleton<ITestMessageTracker, TestMessageTracker>();
        });
        // 其他配置...
    });

// 在消费者中注入并调用跟踪服务
public class YourConsumer : IConsumer<YourTestMessage>
{
    private readonly ITestMessageTracker _tracker;

    public YourConsumer(ITestMessageTracker tracker)
    {
        _tracker = tracker;
    }

    public async Task Consume(ConsumeContext<YourTestMessage> context)
    {
        // 业务处理逻辑
        _tracker.OnMessageReceived();
    }
}

// 测试方法
[Fact]
public async Task SendMessage_ShouldBeConsumed()
{
    var sendEndpoint = await _bus.GetSendEndpoint(new Uri($"queue:your-queue-name"));
    var testMessage = new YourTestMessage();
    var completionSource = new TaskCompletionSource<bool>();

    var tracker = _factory.Services.GetRequiredService<ITestMessageTracker>();
    tracker.MessageReceived += (s, e) => completionSource.TrySetResult(true);

    await sendEndpoint.Send(testMessage);

    Assert.True(await completionSource.Task.WaitAsync(TimeSpan.FromSeconds(10)));
}

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