MassTransit一個優秀的.NET消息（事件）總線框架

1 前置閱讀

在閱讀本文章之前，你可以先閱讀：

• RabbitMQ入門
• 什么是觀察者模式
• 什么是事件總線
• 如何使用RabbitMQ實現事件總線

2 簡介

MassTransit 是一個自由、開源、輕量級的消息總線, 用於使用. NET 框架創建分布式應用程序。MassTransit 在現有消息傳輸上提供了一組廣泛的功能, 從而使開發人員能夠友好地使用基於消息的會話模式異步連接服務。基於消息的通信是實現面向服務的體系結構的可靠和可擴展的方式。

3 使用

首先，將MassTransit的NuGet軟件包安裝到您的應用程序中。

MassTransit
MassTransit.RabbitMQ

然后，創建具體事件源類。

/// <summary>
/// 具體事件源類
/// </summary>
public class SendedEvent
{
    public string Name { get; private set; }
    public SendedEvent(string name)
    {
        Name = name;
    }
}

接着，創建具體事件處理類，默認繼承IConsumer<>接口。

/// <summary>
/// 具體事件處理類顧客A
/// </summary>
public class CustomerASendedEventHandler : IConsumer<SendedEvent>
{
    public Task Consume(ConsumeContext<SendedEvent> context)
    {
        Console.WriteLine($"顧客A收到{context.Message.Name}通知！");
        return Task.CompletedTask;
    }
}
/// <summary>
/// 具體事件處理類顧客B
/// </summary>
public class CustomerBSendedEventHandler : IConsumer<SendedEvent>
{
    public Task Consume(ConsumeContext<SendedEvent> context)
    {
        Console.WriteLine($"顧客B收到{context.Message.Name}通知！");
        return Task.CompletedTask;
    }
}

最后，客戶端調用。

static void Main(string[] args)
{
    var bus = Bus.Factory.CreateUsingRabbitMq(sbc =>
    {
        sbc.Host("rabbitmq://localhost");

        sbc.ReceiveEndpoint("test_queue", ep =>
        {
            ep.Consumer<CustomerASendedEventHandler>();// 訂閱
            ep.Consumer<CustomerBSendedEventHandler>();
        });
    });

    bus.Start();

    var sendedEvent = new SendedEvent("優惠");
    Console.WriteLine($"商店發了{sendedEvent.Name}通知！");
    bus.Publish(sendedEvent);// 發布

    Console.ReadKey(); // press Enter to Stop
    bus.Stop();
}

讓我們來看看輸出結果:

商店發布優惠通知！
顧客A收到優惠通知。
顧客B收到優惠通知。

4 ASP.NET Core 使用

首先，將MassTransit的NuGet軟件包安裝到您的應用程序中。

MassTransit
MassTransit.RabbitMQ
MassTransit.AspNetCore

然后，創建具體事件源類

/// <summary>
/// 具體事件源類
/// </summary>
public class SendedEvent
{
    public string Name { get; private set; }
    public SendedEvent(string name)
    {
        Name = name;
    }
}

接着，創建發布者代碼:

1. 往Startup.ConfigureServices中增加以下代碼

services.AddMassTransit(x =>
{
    x.UsingRabbitMq();
});
services.AddMassTransitHostedService();

2.創建HomeController，發布事件

using EventBus.MassTransit.Events;
using MassTransit;
using Microsoft.AspNetCore.Mvc;
using System.Threading.Tasks;

namespace EventBus.MassTransit.Producer.WebApi01.Controllers
{
    [Route("[controller]")]
    [ApiController]
    public class HomeController : ControllerBase
    {
        private readonly IPublishEndpoint publishEndpoint;
        public HomeController(IPublishEndpoint publishEndpoint)
        {
            this.publishEndpoint = publishEndpoint;
        }

        [HttpPost]
        public async Task<ActionResult> Post()
        {
            await publishEndpoint.Publish<SendedEvent>(new SendedEvent("優惠"));
            return Ok();
        }
    }
}

接着，創建訂閱者代碼:

1. 創建具體事件處理類

/// <summary>
/// 具體事件處理類顧客A
/// </summary>
public class CustomerASendedEventHandler : IConsumer<SendedEvent>
{
    public Task Consume(ConsumeContext<SendedEvent> context)
    {
        Console.WriteLine($"顧客A收到{context.Message.Name}通知！");
        return Task.CompletedTask;
    }
}
/// <summary>
/// 具體事件處理類顧客B
/// </summary>
public class CustomerBSendedEventHandler : IConsumer<SendedEvent>
{
    public Task Consume(ConsumeContext<SendedEvent> context)
    {
        Console.WriteLine($"顧客B收到{context.Message.Name}通知！");
        return Task.CompletedTask;
    }
}

2. 往Startup.ConfigureServices中增加以下代碼

services.AddMassTransit(x =>
{
    x.AddConsumer<CustomerASendedEventHandler>();
    x.AddConsumer<CustomerBSendedEventHandler>();

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.ReceiveEndpoint("event-listener", e =>
        {
            e.ConfigureConsumer<CustomerASendedEventHandler>(context);
            e.ConfigureConsumer<CustomerBSendedEventHandler>(context);
        });
    });
});

services.AddMassTransitHostedService();

最后，執行Post()方法，在訂閱者項目中跟蹤會輸出：

顧客A收到優惠通知。
顧客B收到優惠通知。