MassTransit入門

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收到優惠通知。