MassTransit:消费者消费完所有消息后如何停止总线？

Question

我尝试在使用MassTransit处理完队列中的所有消息后停止总线。我将并发消息限制设置为1，因为我的消费者需要一次处理一条消息。

我尝试过将bus.StopAsync()放在bus.StartAsync后面，如下所示。结果表明，在一条消息被消费后，总线将停止。

总线配置：

IBusControl bus = Bus.Factory.CreateUsingRabbitMq(cfg =>
            {
                IRabbitMqHost host = cfg.Host(new Uri("rabbitmq://localhost"), hostConfigurator =>
                {
                    hostConfigurator.Username("username");
                    hostConfigurator.Password("password");
                });

            cfg.ReceiveEndpoint(host, "MyResult", ep =>
            {
                ep.Bind("MyExchange", s => { s.Durable = true; });

                ep.Consumer<MessageConsumer>(mc =>
                {
                    mc.UseConcurrentMessageLimit(1);
                });
            });
        });

公交车起止时间：

await bus.StartAsync();

await bus.StopAsync();

我的问题是，在队列中的所有消息都被使用完后，如何停止总线。我对MassTransit非常陌生，对调用消费者和停止总线的顺序非常好奇。如果有人能帮上忙我会很感激。谢谢。

Answer 1

在Testing名称空间中，有一个功能用于监视总线上的活动，该功能可用于通知没有消息正在使用(之后，您可以按照您的建议停止总线)。

您可以看到单元测试：https://github.com/MassTransit/MassTransit/blob/v7.0.3/tests/MassTransit.Tests/BusActivityMonitor_Specs.cs#L53

观察者是使用以下命令添加的：

var activityMonitor = bus.CreateBusActivityMonitor(TimeSpan.FromMilliseconds(500));

一旦总线空闲，超时将为真：

var timeout = await activityMonitor.AwaitBusInactivity(TimeSpan.FromSeconds(10));

如果timeout为true，则在指定时间内总线上没有任何活动。