MassTransit和事件与命令发布

Question

我是MassTransit的新手,在我的理解中我想念一些东西.

假设我有一个服务器场,所有节点都可以完成相同的工作.应用程序框架是CQRS的样式.这意味着我有两种基本类型的消息要发布:

• 命令:必须由服务器中的任何一个服务器处理(任何一个服务器空闲)
• 事件:必须由所有服务器处理

我已经构建了一个非常简单的MassTransit原型(一个每X秒发送一次hello的控制台应用程序).

在API中,我可以看到有一种"发布"方法.如何指定它是什么类型的消息(一个与所有服务器)？

如果我看一下"处理程序"配置,我可以指定队列uri.如果我为所有主机指定相同的队列,则所有主机都将获得该消息,但我不能将执行限制为仅一个服务器.

如果我从主机专用队列中侦听,只有一台服务器会处理这些消息,但我不知道如何广播其他类型的消息.

请帮助我理解我所缺少的.

PS:如果它关心,我的消息系统是rabbitmq.

为了测试,我用这个类创建了一个公共类库:

public static class ActualProgram
{
    private static readonly CancellationTokenSource g_Shutdown = new CancellationTokenSource();

    private static readonly Random g_Random = new Random();

    public static void ActualMain(int delay, int instanceName)
    {
        Thread.Sleep(delay);
        SetupBus(instanceName);

        Task.Factory.StartNew(PublishRandomMessage, g_Shutdown.Token);

        Console.WriteLine("Press enter at any time to exit");
        Console.ReadLine();
        g_Shutdown.Cancel();

        Bus.Shutdown();
    }

    private static void PublishRandomMessage()
    {
        Bus.Instance.Publish(new Message
        {
            Id = g_Random.Next(),
            Body = "Some message",
            Sender = Assembly.GetEntryAssembly().GetName().Name
        });

        if (!g_Shutdown.IsCancellationRequested)
        {
            Thread.Sleep(g_Random.Next(500, 10000));
            Task.Factory.StartNew(PublishRandomMessage, g_Shutdown.Token);
        }
    }

    private static void SetupBus(int instanceName)
    {
        Bus.Initialize(sbc =>
        {
            sbc.UseRabbitMqRouting();
            sbc.ReceiveFrom("rabbitmq://localhost/simple" + instanceName);
            sbc.Subscribe(subs =>
            {
                subs.Handler<Message>(MessageHandled);
            });
        });
    }

    private static void MessageHandled(Message msg)
    {
        ConsoleColor color = ConsoleColor.Red;
        switch (msg.Sender)
        {
            case "test_app1":
                color = ConsoleColor.Green;
                break;

            case "test_app2":
                color = ConsoleColor.Blue;
                break;

            case "test_app3":
                color = ConsoleColor.Yellow;
                break;
        }
        Console.ForegroundColor = color;
        Console.WriteLine(msg.ToString());
        Console.ResetColor();
    }

    private static void MessageConsumed(Message msg)
    {
        Console.WriteLine(msg.ToString());
    }
}

public class Message
{
    public long Id { get; set; }

    public string Sender { get; set; }

    public string Body { get; set; }

    public override string ToString()
    {
        return string.Format("[{0}] {1} : {2}" + Environment.NewLine, Id, Sender, Body);
    }
}

我还有3个运行ActualMain方法的控制台应用程序:

internal class Program
{
    private static void Main(string[] args)
    {
        ActualProgram.ActualMain(0, 1);
    }
}

Answer 1

您想要的是竞争消费者(搜索SO以便您可以找到更多信息)使用RabbitMQ可以简化生活,您需要做的就是为您启动的每个消费者指定相同的队列名称,消息将仅由处理其中之一.而不是每次像你一样生成一个唯一的队列.

private static void SetupBus(int instanceName)
{
    Bus.Initialize(sbc =>
    {
        sbc.UseRabbitMqRouting();
        sbc.ReceiveFrom("rabbitmq://localhost/Commands);
        sbc.Subscribe(subs =>
        {
            subs.Handler<Message>(MessageHandled);
        });
    });
}

AFAIK,您需要为命令处理程序提供单独的进程,而不是事件处理程序.所有命令处理程序将从同一队列中接收,所有事件处理程序将从其自己的唯一队列中接收.

另一个难题是如何将消息传入公交车.您仍然可以使用publish for commands,但是如果您错误地配置了使用者,则可以获得多次执行,因为消息将发送给所有使用者,如果您希望保证消息最终在单个队列中,则可以使用"发送"而不是"发布".

     Bus.Instance
         .GetEndpoint(new Uri("rabbitmq://localhost/Commands"))
        .Send(new Message
        {
            Id = g_Random.Next(),
            Body = "Some message",
            Sender = Assembly.GetEntryAssembly().GetName().Name
        });