MassTransit v8内存测试报错Cannot access a disposed object问题求助

我有两个基于RabbitMQ和MassTransit v8的微服务，A向B发送请求，B从数据库取数后返回响应。生产环境运行正常，但用InMemoryTestHarness做集成测试时，抛出错误：

MassTransit.RequestFaultException : The EventBus.Messages.Contracts.Services.Discounts.Currencies.Query.GetCurrencyByName.GetCurrencyByNameRequest request faulted: Cannot access a disposed object。

相关代码片段 #

请求/响应消息定义

// 请求消息
public class GetCurrencyByNameRequest
{
    public string Name { get; set; }
}

// 响应消息
public class GetCurrencyByNameResponse
{
    public string Id { get; set; } 
    public string Name { get; set; } 
}

消费者代码

public class GetCurrencyByNameConsumer : IConsumer<GetCurrencyByNameRequest>
{
    private readonly IUnitOfWork _uow;

    public GetCurrencyByNameConsumer(IUnitOfWork uow)
    {
        _uow = uow;
    }
    public async Task Consume(ConsumeContext<GetCurrencyByNameRequest> context)
    {
        var filter = Builders<Currency>.Filter.Eq(x => x.CurrencyName, new CurrencyName(context.Message.Name));
        var currency = await _uow.GenericRepository<Currency>()
            .GetSingleDocumentByFilterAsync(filter, context.CancellationToken);
        if (currency == null)
        {
            await context.RespondAsync(new GetCurrencyByNameResponse
            {
                Id = string.Empty,
                Name = string.Empty
            });
        }
        else
        {
            await context.RespondAsync(new GetCurrencyByNameResponse
            {
                Id = currency.Id,
                Name = currency.CurrencyName
            });
        }
    }
}

微服务A MassTransit配置

services.AddMassTransit(x =>
{
    var entryAssembly = AppDomain.CurrentDomain.GetAssemblies()
        .FirstOrDefault(x => x.FullName.Contains("Payment.Application"));
    x.SetKebabCaseEndpointNameFormatter();
    x.AddConsumers(entryAssembly);
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(configuration["RabbitMqSettings:Host"], configuration["RabbitMqSettings:VirtualHost"], h =>
        {
            h.Username(configuration["RabbitMqSettings:Username"]);
            h.Password(configuration["RabbitMqSettings:Password"]);
        });
        cfg.UseDelayedRedelivery(r => r.Intervals(TimeSpan.FromMinutes(5), TimeSpan.FromMinutes(15), TimeSpan.FromMinutes(30)));
        cfg.UseMessageRetry(r => r.Immediate(5));
        cfg.UseInMemoryOutbox();
        cfg.ConfigureEndpoints(context);
    });
});

微服务B MassTransit配置

services.AddMassTransit(x =>
{
    var entryAssembly = AppDomain.CurrentDomain.GetAssemblies()
        .FirstOrDefault(x => x.FullName.Contains("Discount.Application"));
    x.SetKebabCaseEndpointNameFormatter();
    x.AddConsumers(entryAssembly);
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(configuration["RabbitMqSettings:Host"], configuration["RabbitMqSettings:VirtualHost"], h =>
        {
            h.Username(configuration["RabbitMqSettings:Username"]);
            h.Password(configuration["RabbitMqSettings:Password"]);
        });
        cfg.UseDelayedRedelivery(r => r.Intervals(TimeSpan.FromMinutes(5), TimeSpan.FromMinutes(15), TimeSpan.FromMinutes(30)));
        cfg.UseMessageRetry(r => r.Immediate(5));
        cfg.UseInMemoryOutbox();
        cfg.ConfigureEndpoints(context);
    });
});

内存测试Harness配置

public static InMemoryTestHarness TestHarness { get; set; } = default!;
public static IBus TestHarnessBus { get; set; } = default!;
public static async Task UseHarnessAsync()
{
    await using var provider = new ServiceCollection()
            .AddScoped<IUnitOfWork, UnitOfWork>()
            .AddMassTransitInMemoryTestHarness(x =>
                    {
                        x.SetKebabCaseEndpointNameFormatter();
                        x.AddConsumer<GetCurrencyByNameConsumer>();
                        x.AddConsumerTestHarness<GetCurrencyByNameConsumer>();
                    })
            .AddGenericRequestClient()
            .BuildServiceProvider(true);

    TestHarness = provider.GetRequiredService<InMemoryTestHarness>();
    await TestHarness.Start().ConfigureAwait(false); 
    TestHarnessBus = provider.GetRequiredService<IBus>();
}

集成测试代码

[Test]
public async Task CreatePackageCommand_InsertToDb_ReturnPackage()
{
    IRequestClient<GetCurrencyByNameRequest>? client = TestHarnessBus.CreateRequestClient<GetCurrencyByNameRequest>();
    await Task.Delay(20000);
    var req = new GetCurrencyByNameRequest
    {
        Name = "Dollar"
    };
    // 错误发生在此行
    var response = await client.GetResponse<GetCurrencyByNameResponse>(req);

    (await TestHarness.Sent.Any<GetCurrencyByNameRequest>()).Should().BeTrue();
    (await TestHarness.Consumed.Any<GetCurrencyByNameResponse>()).Should().BeTrue();

    var consumerHarness = GetConsumerTestHarness<GetCurrencyByNameConsumer>();

    (await consumerHarness.Consumed.Any<GetCurrencyByNameRequest>()).Should().BeTrue();
    var command = new CreatePackageCommand
    {
        IsActive = true,
        Description = new Translation
        {
            Arabic = StringHelper.RandomString(10),
            English = StringHelper.RandomString(10),
            Persian = StringHelper.RandomString(10),
        },
        Sort = 1,
        Name = new Translation
        {
            Arabic = StringHelper.RandomString(10),
            English = StringHelper.RandomString(10),
            Persian = StringHelper.RandomString(10),
        },
        Price = 999,
        CurrencyName = "Dollar"
    };
    var packageId = await SendAsync(command);
}

问题根源与修复方案 #

核心问题：ServiceProvider 提前释放 #

测试Harness配置中，provider 用 await using 声明，导致 UseHarnessAsync 方法执行完毕后，容器被自动释放。后续测试代码中用到的 TestHarnessBus、TestHarness 以及消费者依赖的 IUnitOfWork 都绑定在已释放的容器上，因此消费请求时触发「已释放对象访问」错误。

修复步骤 #

1. 延长ServiceProvider生命周期
   将 provider 提升为静态变量，避免方法执行完毕后被自动释放：

   public static InMemoryTestHarness TestHarness { get; set; } = default!;
   public static IBus TestHarnessBus { get; set; } = default!;
   public static IServiceProvider? Provider { get; set; } // 新增静态容器变量
   
   public static async Task UseHarnessAsync()
   {
       Provider = new ServiceCollection()
               .AddScoped<IUnitOfWork, UnitOfWork>()
               .AddMassTransitInMemoryTestHarness(x =>
                       {
                           x.SetKebabCaseEndpointNameFormatter();
                           x.AddConsumer<GetCurrencyByNameConsumer>();
                           x.AddConsumerTestHarness<GetCurrencyByNameConsumer>();
                       })
               .AddGenericRequestClient()
               .BuildServiceProvider(true);
   
       TestHarness = Provider.GetRequiredService<InMemoryTestHarness>();
       await TestHarness.Start().ConfigureAwait(false); 
       TestHarnessBus = Provider.GetRequiredService<IBus>();
   }
   
   // 测试结束后手动释放资源
   public static async Task StopHarnessAsync()
   {
       await TestHarness.Stop().ConfigureAwait(false);
       if (Provider is IAsyncDisposable asyncDisposable)
           await asyncDisposable.DisposeAsync();
       else
           Provider?.Dispose();
   }
   

2. 在测试类中管理Harness生命周期
   在测试的初始化与清理阶段分别启动、停止Harness：

   [OneTimeSetUp]
   public async Task Setup()
   {
       await YourHarnessClass.UseHarnessAsync();
   }
   
   [OneTimeTearDown]
   public async Task TearDown()
   {
       await YourHarnessClass.StopHarnessAsync();
   }
   

3. 移除无效延迟
   测试代码中的 await Task.Delay(20000) 完全不必要，InMemoryTestHarness启动后即可使用，直接删除该行。

4. 修正断言逻辑
   原测试中对响应消息的断言错误，GetCurrencyByNameResponse 是响应，不会被 TestHarness.Consumed 捕获，应改为检查请求是否被消费者处理：

   var consumerHarness = TestHarness.GetConsumerTestHarness<GetCurrencyByNameConsumer>();
   (await consumerHarness.Consumed.Any<GetCurrencyByNameRequest>()).Should().BeTrue();
   

额外检查点 #

• 确保 IUnitOfWork 及其依赖（如MongoDB客户端）在测试容器中生命周期配置正确，避免提前释放；
• 建议将测试中的真实数据库替换为内存数据库（如MongoDB内存实例），消除外部资源依赖。

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