如何使用 .NET Core 7 控制台向 MassTransit 消费者发送数据

Question

我的解决方案中有两个项目，一个是控制台，另一个是 API。在 API 项目中，我按如下方式设置 MassTransit：

using MassTransit;
using Notification;

var builder = WebApplication.CreateBuilder(args);

// Add services to the container.

builder.Services.AddControllers();

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<NotificationRequestConsumer>();
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(new Uri("rabbitmq://localhost"), h =>
        {
            h.Username("guest");
            h.Password("guest");
        });
        cfg.ReceiveEndpoint("notificationQueue", ep =>
        {
            ep.PrefetchCount = 16;
            ep.UseMessageRetry(r => r.Interval(2, 100));
            ep.ConfigureConsumer<NotificationRequestConsumer>(context);
        });
    });
});

var app = builder.Build();

以及

NotificationRequestConsumer

如下：

public class NotificationRequestConsumer : IConsumer<INotificationRequest>
{
    public Task Consume(ConsumeContext<INotificationRequest> context)
    {
        Console.WriteLine(context.Message.Event);

       return Task.CompletedTask;
    }
}

现在我需要向该消费者发送消息，我该怎么做？

我设置

Program.cs

如下：

class Program
{
    static void Main(string[] args)
    {
        CreateHostBuilder(args).Build();
    }

    public static IHostBuilder CreateHostBuilder(string[] args) =>
    Microsoft.Extensions.Hosting.Host.CreateDefaultBuilder(args)
        .ConfigureServices((hostContext, services) =>
        {
            services.AddMassTransit(x =>
            {
                x.UsingRabbitMq((context, config) =>
                {
                    config.Host(new Uri("rabbitmq://localhost"), h =>
                    {
                        h.Username("guest");
                        h.Password("guest");
                    });
                });
            });
        });
}

但我一直坚持向该消费者发送消息。请帮我解决这个问题。

Answer 1

首先确保消息契约 (

INotificationRequest

) 是在控制台应用程序和 API 项目均可访问的共享库中定义的。在控制台应用程序中，您需要发布一个符合

INotificationRequest

的消息实例合同。这是示例代码：

class Program
{
    static async Task Main(string[] args)
    {
        var host = CreateHostBuilder(args).Build();

        var publishEndpoint = host.Services.GetRequiredService<IPublishEndpoint>();

        var message = new NotificationRequest
        {
            Event = "Sample Event"
        };

        await publishEndpoint.Publish<INotificationRequest>(message);

        Console.WriteLine("Message published");
    }

    public static IHostBuilder CreateHostBuilder(string[] args) =>
        Host.CreateDefaultBuilder(args)
            .ConfigureServices((hostContext, services) =>
            {
                services.AddMassTransit(x =>
                {
                    x.UsingRabbitMq((context, config) =>
                    {
                        config.Host(new Uri("rabbitmq://localhost"), h =>
                        {
                            h.Username("guest");
                            h.Password("guest");
                        });
                    });
                    x.AddPublishMessageScheduler();
                });
            });
}