RabbitMQ 与 ASP.NET Core – 与 MassTransit 的微服务通信

public void ConfigureServices(IServiceCollection services)
{

    services.AddMassTransit(x =>
    {
        x.AddBus(provider => Bus.Factory.CreateUsingRabbitMq(config =>
        {
            config.Host(new Uri(RabbitMqConsts.RabbitMqRootUri), h =>
            {
                h.Username(RabbitMqConsts.UserName);
                h.Password(RabbitMqConsts.Password);
            });
        }));
    });
    services.AddMassTransitHostedService();
    services.AddControllers();
    services.AddSwaggerGen(c =>
    {
        c.SwaggerDoc("v1", new OpenApiInfo { Title = "Microservice.Todo.Publisher", Version = "v1" });
    });
}

在我们的发布者控制器中，我们将创建一个发布方法来发布消息。

TodoController.cs

[HttpPost]
public async Task<IActionResult> CreateTicket(Todo todoModel)
{
    if (todoModel is not null)
    {
        Uri uri = new Uri(RabbitMqConsts.RabbitMqUri);
        var endPoint = await _bus.GetSendEndpoint(uri);
        await endPoint.Send(todoModel);
        return Ok();
    }
    return BadRequest();
}

Notification类库

为了测试，我们将看到一个控制台应用程序。在下面的代码片段中，我们将在 Program.cs 文件中注册消费，并在 TodoConsumerNotification.cs 文件中编写消费方法来显示消息。

static void Main(string[] args)
{
    Console.Title = "Notification";
    var bus = Bus.Factory.CreateUsingRabbitMq(cfg =>
    {
        cfg.Host(new Uri(RabbitMqConsts.RabbitMqRootUri), h =>
        {
            h.Username(RabbitMqConsts.UserName);
            h.Password(RabbitMqConsts.Password);
        });
        cfg.ReceiveEndpoint("todoQueue", ep =>
        {
            ep.PrefetchCount = 16;
            ep.UseMessageRetry(r => r.Interval(2, 100));
            ep.Consumer<TodoConsumerNotification>();
        });

    });

    bus.StartAsync();
    Console.WriteLine("Listening for Todo registered events.. Press enter to exit");
    Console.ReadLine();
    bus.StopAsync();
}

public class TodoConsumerNotification : IConsumer<Todo>
{
    public async Task Consume(ConsumeContext<Todo> context)
    {
        await Console.Out.WriteLineAsync($"Notification sent: todo id {context.Message.Id}");
    }
}

测试我们的应用程序

方案 1

保持Publisher和Consumers/Notification作为多个启动项目运行。

通过 Swagger 调用 post 方法 API/todo，如下图所示，

输出

方案 2

保持 Publisher 设置为启动项目，Consumer/Notification 运行为离线。

通过 Swagger 调用 post 方法 API/todo，如下图所示，

您可以看到我们的 Queue 有 1 个新的 Message Ready 尚未交付。它将消息保存在内存中，直到连接消费者。

现在让我们的 Consumer 上线，我们可以看到消息已传递，Queue 为 0。

总结

在本文中，我们介绍了消息代理、RabbitMQ、优势、使用 MassTransit 将 RabbitMQ 与 ASP.NET Core 集成。我们还构建了一个小型原型应用程序来通过 RabbitMQ 服务器发送数据。从 GitHub 下载源代码