使用MassTransit发布消息的方法是通过交换器和路由键来实现的。

Question

我已经花了几个星期的时间研究 MassTransit，对其可能性很感兴趣。然而，我似乎还没有完全理解这些概念。

期望行为 我想将消息发布到一个“直接”交换机，该交换机绑定到两个不同的队列以执行其他活动。

当我尝试使用 MassTransit 实现相同的逻辑以获得更好的可扩展性时，我发现 MassTransit 基于队列名称创建自己的交换机，并采用扇形类型。

发布消息通过交换机和路由键的经典代码

using (var connection = factory.CreateConnection())
        {
            using (var channel = connection.CreateModel())
            {
                channel.ExchangeDeclare(exchange, "direct");

                var body = Encoding.UTF8.GetBytes(message);

                channel.BasicPublish(exchange, routingKey, null, body);
                Console.WriteLine(" [x] Sent {0}", message);
            }
        }

是否有办法在MassTransit中配置直接或主题交换并使用路由键？

- Syed

Answer 1

以下代码完成相同的工作，但增加了一个扇出交换机：

测试消息（直连交换机）-> 测试消息队列（扇出交换机）-> 测试消息队列（队列）

var bus = Bus.Factory.CreateUsingRabbitMq(cfg =>
{
    var host = cfg.Host(new Uri("rabbitmq://localhost"), h =>
    {
        h.Username("guest");
        h.Password("guest");
    });
    cfg.Send<TestMessage>(x => { x.UseRoutingKeyFormatter(context => "routingKey"); });
    cfg.Message<TestMessage>(x => x.SetEntityName("TestMessage"));
    cfg.Publish<TestMessage>(x => { x.ExchangeType = ExchangeType.Direct; });
    cfg.ReceiveEndpoint(host, "TestMessage_Queue", e =>
    {
        e.BindMessageExchanges = false;
        e.Consumer<UpdateCustomerConsumer>();
        e.Bind("TestMessage", x =>
        {
            x.ExchangeType = ExchangeType.Direct;
            x.RoutingKey = "routingKey";
        });
    });
});

bus.Start();

- Jan Muncinsky

---

这是哪个版本？版本 5.5.1 没有 Send<T> 或 Publish<T>。 - Andre

Answer 2

这不是MassTransit支持的方案。 MassTransit将始终创建一个扇出队列。如果您自己管理拓扑结构，则可以使用IEndpoint.Send直接将消息发送到您创建的交换机。但在这种情况下，您会失去MT提供的许多功能。

我也不确定在这种情况下“更好的可伸缩性”是什么意思。与直接交换相比，扇出交换（在大多数情况下）表现更好，因为无需处理路由逻辑。

也许如果您在MassTransit邮件列表上澄清了您的性能问题，我们可以在那里为您提供更多帮助。

- Travis

---

感谢您的评论。我的进程需要多个发布者并行发布消息，因此考虑使用“直接”或“主题”交换，但它们在MassTransit中不受支持。您有任何关于处理此问题的最佳方法的建议吗？或者您是否更喜欢为此进程使用多个队列。 此外，如果您能指导我了解MassTransit的架构，那将非常有帮助。 - Syed

我不太确定你想做什么。并行发布消息对我来说没有太多意义。使用MT 3.0，您可以进行异步发布。但是，通过消息传递，一切都更或多或少地是并行的。如果一个线程正在发布，那么另一个线程也可以这样做。消费也是如此。 - Travis

基本上，忘掉 RMQ 中的拓扑结构，只需使用类型（最好是接口）定义消息契约，让 MT 将其全部绑定在一起。 - Travis

Answer 3

MassTransit处理发布和订阅而不使用主题交换，而是为正在发布的消息类型创建交换，并将这些交换绑定到消费队列。

使用路由键方法在RabbitMQ中效率较低，它更喜欢使用交换机结构以实现消息路由的简单性（无需维护哈希表）。

只需定义您的命令和/或事件类型，然后发送或发布这些消息，让消费者完成其工作，而无需处理交换和路由键。

- Chris Patterson

---

创建额外的类型虽然非常清晰，但可能会很繁琐。一种诱人的方法是让您的总线甚至不知道交换了哪些确切的类型，因为您将所有内容都包装在一个通用的“类似包装器”的类型中（可能会丢失静态类型，因为只有单个包装器类型）。但是，单个类似包装器类型的缺点是，在技术上，存在从单个交换到所有订阅者的扇出，因此所有订阅者都检索所有消息，必须在处理程序内部进行过滤。 - Wiktor Zychla

这就是路由键方法变得方便的地方。通过为每个包装消息设置显式键和将路由键映射到队列，您可以精确控制订阅。幸运的是，正如@jan-mucinsky在其他答案中提到的那样，这似乎在MT中起作用。我不是为这种模型进行宣传，我只是说这在技术上是可能的，并且有其合理性。 - Wiktor Zychla

@WiktorZychla，你能想出如何实现这个吗？ - Aranha

@Aranha：没错，我在那上面写了一篇博客（https://www.wiktorzychla.com/2019/03/）。 - Wiktor Zychla