如何通过 MassTransit 和 RabbitMQ 发送各种命令类型？答案

【问题标题】：How to send various command type by MassTransit and RabbitMQ?如何通过 MassTransit 和 RabbitMQ 发送各种命令类型？
【发布时间】：2017-09-11 11:01:45
【问题描述】：

我是使用消息代理的初学者。
我们的票务服务有多个子服务。主管服务在 Web API 的帮助下获取请求并将它们发送到子服务。
任何请求都有一个标头，用于检测命令类型（例如 Reserve、Refund、Availability 等）。我们使用 json 来序列化对象。
现在，如何通过 MassTransit 从发布者（例如我们的主管系统）发送各种消息类型（不同的对象），让消费者可以轻松使用它？
一般来说，MassTransit 和 rabbitMQ 是否可以发送各种类型的消息？
每个消费者只有一个队列用于处理接收到的消息。

谢谢

更新

https://dotnetcodr.com/2016/08/02/messaging-with-rabbitmq-and-net-review-part-1-foundations-and-terminology/

我阅读了这篇文章适合开始使用 MassTransit 进行消息传递，但没有看到在这些资源和其他资源上使用各种消息类型的任何示例：

我有多个命令，需要各种消息类型来发送，但在示例中只使用如下消息类型：

发件人

    private static void RunMassTransitPublisherWithRabbit()
    {
        string rabbitMqAddress = "rabbitmq://localhost:5672/Ticket";
        string rabbitMqQueue = "mycompany.domains.queues";
        Uri rabbitMqRootUri = new Uri(rabbitMqAddress);

        IBusControl rabbitBusControl = Bus.Factory.CreateUsingRabbitMq(rabbit =>
        {
            rabbit.Host(rabbitMqRootUri, settings =>
            {
                settings.Password("Kalcho^Milano");
                settings.Username("ticketadmin");
            });
        });

        Task<ISendEndpoint> sendEndpointTask = rabbitBusControl.GetSendEndpoint(new Uri(string.Concat(rabbitMqAddress, "/", rabbitMqQueue)));
        ISendEndpoint sendEndpoint = sendEndpointTask.Result;

        Task sendTask = sendEndpoint.Send<IRegisterCustomer>(new
        {
            Address = "New Street",
            Id = Guid.NewGuid(),
            Preferred = true,
            RegisteredUtc = DateTime.UtcNow,
            Name = "Nice people LTD",
            Type = 1,
            DefaultDiscount = 0
        });
        Console.ReadKey();
    }

接收者

        private static void RunMassTransitReceiverWithRabbit()
    {
        IBusControl rabbitBusControl = Bus.Factory.CreateUsingRabbitMq(rabbit =>
        {
            IRabbitMqHost rabbitMqHost = rabbit.Host(new Uri("rabbitmq://localhost:5672/Ticket"), settings =>
            {
                settings.Password("Kalcho^Milano");
                settings.Username("ticketadmin");
            });

            rabbit.ReceiveEndpoint(rabbitMqHost, "mycompany.domains.queues", conf =>
            {
                conf.Consumer<RegisterCustomerConsumer>();
            });
        });

        rabbitBusControl.Start();
        Console.ReadKey();

        rabbitBusControl.Stop();
    }

IRegisterCustomer 是一个接口，我只能在rabbit.ReceiveEndpoint 中获取消息内容并转换为可用对象。

现在，如何使用IReserveTicket、IRefundTicket、IGetAvailability等各种消息类型来发送和接收消息？

再次感谢

【问题讨论】：

您阅读过文档吗？ masstransit-project.com/MassTransit/usage/…
说实话，我不明白你的问题。为什么不能创建与消息一样多的消费者并发送这些消息？有什么问题？
我认为不能多于一种类型的消息发送给消费者，所以设计了一个复杂的结构，一个消息包含系统中消息的所有细节，这个想法是错了。
每个消费者都在实现IConsumer<T>，其中T 是消息类型。所以是的，一个消费者只能消费一种消息类型。但是您可以拥有任意数量的消费者，每个端点可以拥有任意数量的消费者，并且可以拥有任意数量的端点。

标签： c# rabbitmq microservices masstransit

【解决方案1】：

如果您向端点添加更多消费者，像这样

rabbit.ReceiveEndpoint(rabbitMqHost, "mycompany.domains.queues", conf =>
{
    conf.Consumer<RegisterCustomerConsumer>();
    conf.Consumer<ReserveTicketConsumer>();
    conf.Consumer<RefundTicketConsumer>();
});

并发送类似的消息

await endpoint.Send<IReserveTicket>(new { TickedId = 123 });

它会起作用的。

上述解决方案假设您没有负载过重，尤其是负载不均，您会收到数百万条一种类型的消息，而可能是数百种其他类型的消息。将它们全部放在一个端点会造成消费失衡，因为所有这些消费者只有一个队列。在这种情况下，没有什么能阻止您根据需要定义任意数量的端点，每个端点都应该有一个单独的队列。例如：

cfg.ReceiveEndpoint(rabbitMqHost, "mycompany.domains.lowvolume", 
    c =>
    {
        c.Consumer<RegisterCustomerConsumer>();
        c.Consumer<RefundTicketConsumer>();
    });
cfg.ReceiveEndpoint(rabbitMqHost, "mycompany.domains.highvolume", 
    c => c.Consumer<ReserveTicketConsumer>();

请记住，由于您有不同的队列，因此您需要使用这些地址来获取发送端点。

【讨论】：