添加消费者时的大批量运输

Question

[我试图理解使用MassTransit时发生的情况，而不是完全依赖于此抽象层库，而是真正地了解和了解幕后创建的内容及其背后的原因。

在我的应用程序中，我通过以下方式注册消费者：

container.AddMassTransit(x =>
{
    x.AddBus(context => Bus.Factory.CreateUsingRabbitMq(cfg =>
    {
        var host = cfg.Host(configurationProvider.RabbitHostName);

        x.AddConsumer<FactAddedHandler>();
        x.AddConsumer<FactAddedOrderHandler>();
        x.AddConsumer<FactCategoryHandler>();

        cfg.ConfigureEndpoints(container);
    }));

});

跟随我，到目前为止，我们有这种情况：

[当我检查通过创建哪些队列时

rabbitmqctl list_queues名称消息的使用者

我看到：

FactAddedHandler     0       1
FactAddedOrderHandler        0       1
FactCategoryHandler  0       1

这确实使我相信每个特定消费者都有一个队列。因此，我们有这种情况：

让我们看看如何定义这些处理程序：

internal class FactAddedHandler : IConsumer<FactAddedIntegrationEvent>
{
    //
}


internal class FactAddedOrderHandler : IConsumer<FactAddedIntegrationEvent>
{
    //
}


internal class FactCategoryHandler : IConsumer<FactCategoryIntegrationEvent>
{
    //
}

因此，前两个处理程序（FactAddedOrderHandler和FactAddedOrderHandler）订阅同一事件（FactAddedIntegrationEvent），而另一个（FactCategoryHandler **）订阅另一个事件（* * FactCategoryIntegrationEvent）。

因此，我希望在前2个队列的顶部至少有一个fanout交换，该队列广播FactAddedIntegrationEvent。因此，让我们通过以下方式查看交易情况：

rabbitmqctl list_exchanges名称类型

结果是：

FactCategoryHandler     fanout
FactAddedOrderHandler      fanout
FactAddedHandler   fanout

IntegrationEvents:FactAddedIntegrationEvent  fanout
IntegrationEvents:FactCategoryIntegrationEvent    fanout

所以..我期望的是，[[IntegrationEvents：FactAddedIntegrationEvent是广播FactAddedIntegrationEvent的扇出交换机。

我也相信，默认情况下，masstransit会创建另一个交易所

IntegrationEvents：FactCategoryIntegrationEvent

，即使在我的案例中只有一个消费者，也很容易将其他使用者添加到同一事件中。因此，我们最终还是遇到了这种情况：

我不理解，并希望解释的是，创建了其余三个交易所的原因。他们的作用是什么？他们为什么在那里？预先感谢！

Answer 1

对于接收端点，MassTransit会创建队列和具有相同名称的匹配交换。队列绑定到匹配的交换。

使用者在接收端点上消耗的消息类型用于声明交换（如您在上面所示），并且匹配的交换绑定到那些消息类型交换。所有这些交换都是

fanout交换。

生成的拓扑通过匹配的交换路由所有消息类型，然后将其路由到队列。

此示例在documentation中进行了说明。

为什么进行匹配交换？两个答案。

首先是简单的答案，它允许使用路由密钥将消息发送到交换机。使用RabbitMQ发送到队列需要使用队列名称作为路由密钥进行空交换。

其次，它允许将其他队列绑定到匹配的交换机，以进行故障排除。这包括设置窃听，以保留发送到端点的所有消息的副本（直接发送或通过已发布消息类型交换发送给端点）。