重复消息 Masstransit rabbitmq

Question

我正在尝试为我的应用程序实现心跳功能，因此我正在尝试从 masstransit rabbitmq 实现重复消息功能。我试图在 masstransit 网站上提供的样本上实施它。这是所有代码。

namespace MasstransitBasicSample
{
    using System;
    using System.Threading.Tasks;
    using MassTransit;
    using MassTransit.Scheduling;

    class Program
    {
        static void Main(string[] args)
        {
            var bus = Bus.Factory.CreateUsingRabbitMq(sbc =>
            {
                var host = sbc.Host(new Uri("rabbitmq://localhost"), h =>
                {
                    h.Username("guest");
                    h.Password("guest");
                });

                sbc.UseMessageScheduler(new Uri("rabbitmq://localhost/quartz"));

                sbc.ReceiveEndpoint(host, "test_queue", ep =>
                {
                    ep.Handler<YourMessage>(context =>
                    {
                        return Console.Out.WriteLineAsync($"Received: {context.Message.Text}");
                    });
                    ep.Handler<PollExternalSystem>(context =>
                    {
                        return Console.Out.WriteLineAsync($"Received: {context.Message}");
                    });
                });
            });

            bus.Start();
            SetRecurring(bus);

            Console.WriteLine("Press any key to exit");
            Console.ReadKey();

            bus.Stop();
        }

        private static async Task SetRecurring(IBusControl bus)
        {
            var schedulerEndpoint = await bus.GetSendEndpoint(new Uri("rabbitmq://localhost/quartz"));

            var scheduledRecurringMessage = await schedulerEndpoint.ScheduleRecurringSend(
               new Uri("rabbitmq://localhost/test_queue"), new PollExternalSystemSchedule(), new PollExternalSystem());
        }
    }

    public class YourMessage { public string Text { get; set; } }

    public class PollExternalSystemSchedule : DefaultRecurringSchedule
    {
        public PollExternalSystemSchedule()
        {
            CronExpression = "* * * * *"; // this means every minute
        }
    }

    public class PollExternalSystem { }
}

我在我的 rabbitmq 队列中创建了一个名为 quartz 的队列。

当我运行应用程序将一条消息发送到石英队列并且该消息只是停留在那里时，它不会进入测试队列。

根据 cron 表达式，我还希望在一分钟后将另一条消息发送到石英队列，但这也没有发生。

我的设置有误吗？

如有任何帮助，我们将不胜感激。

Answer 1

您需要运行监听 rabbitmq://localhost/quartz 的调度服务，您的消息将发送到该位置。

documentation page 说：

There is a standalone MassTransit service, MassTransit.QuartzService, which can be installed and used on servers for this purpose. It is configured via the App.config file and is a good example of how to build a standalone MassTransit service.

或者，您可以使用内存中调度在同一进程中托管 Quartz 调度，described here，配置如下：

var busControl = Bus.Factory.CreateUsingRabbitMq(cfg =>
{
    var host = cfg.Host(new Uri("rabbitmq://localhost/"), h =>
    {
        h.Username("guest");
        h.Password("guest");
    });

    cfg.UseInMemoryScheduler();
});

重复消息 Masstransit rabbitmq

Recurring message Masstransit rabbitmq

masstransit