.Net Core 5.x Api開發筆記 -- 消息隊列RabbitMQ實現事件總線EventBus(二)

上一節說了事件總線

本節在消息隊列中實現事件處理：.Net Core 5.x Api開發筆記 -- 消息隊列RabbitMQ實現事件總線EventBus(一)

既然是消息隊列，就需要有生產者和消費者(訂閱)

public interface IMessageQueue
{
    /// <summary>
    /// 發布消息
    /// </summary>
    void Publish<T>(T message, string exchangeName = "default", string queueName = "") where T : class;

    /// <summary>
    /// 訂閱消息
    /// </summary>
    void Consume<T>(Func<T, Task<bool>> func, string queueName = "");
}

生產者端實現發布消息接口:

public class RabbitMQPublishClient : IMessageQueue
{
    private static ConcurrentDictionary<string, string> QueueDic = new ConcurrentDictionary<string, string>();
    public RabbitMQPublishClient(){}
    
    /// <summary>
    /// 發布消息
    /// </summary>
    /// <typeparam name="T"></typeparam>
    /// <param name="message">消息實體</param>
    /// <param name="exchangeName">交換機名稱（默認default）</param>
    /// <param name="queueName">隊列名稱（默認類名）</param>
    public void Publish<T>(T message, string exchangeName = "default", string queueName = "") where T : class
    {
        using (var conn = GetConnection())
        {
            using (var channel = conn.CreateModel())
            {
                if (!QueueDic.ContainsKey(queueName) || QueueDic[queueName] != exchangeName)
                {
                    CreateQueue(channel, exchangeName, queueName, true);
                    QueueDic.TryAdd(queueName, exchangeName);
                }
                var props = channel.CreateBasicProperties();
                props.Persistent = true;  //消息持久化
                props.DeliveryMode = 2;   //消息持久化
                props.CorrelationId = Guid.NewGuid().ToString();

                string content = JsonConvert.SerializeObject(message);
                var body = Encoding.UTF8.GetBytes(content);
                channel.BasicPublish(exchange: exchangeName, routingKey: queueName, basicProperties: props, body: body);
            }
        }
    }

    /// <summary>
    /// 創建隊列，綁定到交換機
    /// </summary>
    private void CreateQueue(IModel channel, string exchangeName, string queueName, bool isDurable = true)
    {
        channel.ExchangeDeclare(exchangeName, ExchangeType.Direct, durable: isDurable, autoDelete: false, arguments: null);
        channel.QueueDeclare(queueName, durable: isDurable, exclusive: false, autoDelete: false, arguments: null);
        channel.QueueBind(queueName, exchangeName, routingKey: queueName);
    }
}

消費者(訂閱者)實現訂閱接口

public class RabbitMQConsumerClient : IMessageQueue
{
    public RabbitMQConsumerClient(){}

    /// <summary>
    /// 訂閱消息
    /// </summary>
    public void Consume<T>(Func<T, Task<bool>> func, string queueName = "")
    {
        string exchangeName = exchangeName.Equals("default") ? $"{customKey}.default" : $"{customKey}.{queueName}";

        var collection = GetConnection();
        var channel = collection.CreateModel();
        channel.ExchangeDeclare(exchangeName, "direct", durable: true, autoDelete: false, arguments: null);
        channel.QueueDeclare(queue: queueName, durable: true, exclusive: false, autoDelete: false, arguments: null);
        channel.QueueBind(queueName, exchangeName, routingKey: queueName, arguments: null);

        //消費事件
        var consumer = new EventingBasicConsumer(channel);
        consumer.Received += async (sender, e) =>
        {
            try
            {
                string content = Encoding.UTF8.GetString(e.Body.ToArray());
                T message = JsonConvert.DeserializeObject<T>(content);
                bool isSuccess = await func.Invoke(message);    //執行func委托  這里才是真正執行傳入的事件處理程序
                if (isSuccess)
                {
                    channel.BasicAck(e.DeliveryTag, false);     //手動確認消息消費成功
                }
                else
                {
                   channel.BasicReject(e.DeliveryTag, true);    //手動打回隊列，下次重新消費
                }
            }
            catch (Exception ex)
            {
                channel.BasicAck(e.DeliveryTag, false);
            }
        };

        channel.BasicQos(0, 1, false);  //限制同時接收消息數量為1，true是將限制應用於channel級，false是將限制應用於consumer級
        channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);   //訂閱消息   autoAck: false 表示手動確認消息
    }
}

上邊都是平時很常用的功能邏輯，接下來是才是重點！！！

1，事件總線有了

2，生產者訂閱者也有了

接下來還需要做至少3件事：

1，創建事件處理程序，就是真正干活的

2，注冊消費者、注冊事件處理程序、綁定事件源和事件處理程序（在訂閱端）

3，訂閱端啟動消費者訂閱監控，並觸發事件處理程序

4，發布消息測試

-----------------------------------------------------------------

1，創建事件處理程序

/// <summary>
/// 事件處理程序，真正干活的是我
/// 繼承IEventHandler<T>接口，其中T必須用EventWithData<TData>類型，因為EventWithData<TData>繼承了EventBase類
/// </summary>
public class Person_EventHandler : IEventHandler<EventWithData<Person>>
{
    //最終干活的
    public async Task HandleEvent(EventWithData<Person> eventWithData)
    {
        var data = eventWithData.Data;
        if (data == null)
        {
            Console.WriteLine("么有數據呀!");
        }
        else
        {
            try
            {
                Console.WriteLine($"{DateTime.Now}-------" + JsonConvert.SerializeObject(data));
            }
            catch (Exception ex)
            {
                Console.WriteLine("異常：" + ex.Message);
            }
        }
    }
}

2，注冊消費者、注冊事件處理程序、綁定事件源和事件處理程序（在訂閱端），這一步在 Startup 中實現！！！

public IServiceCollection ServiceCollection { get; private set; }
public void ConfigureServices(IServiceCollection services)
{
    ServiceCollection = services;    //將services傳遞給ServiceCollection
    services.AddControllersWithViews();

    //注冊消費者
    services.AddSingleton<IMessageQueue, RabbitMQConsumerClient>();
    //注冊事件處理程序，綁定事件源：IEventHandler<EventWithData<Person>>和事件處理程序：Person_EventHandler的關系
    services.AddSingleton<Person_EventHandler>();
}

public void Configure(IApplicationBuilder app, IWebHostEnvironment env)
{
    //綁定事件源和事件處理程序的映射關系  
    //這里跟上邊的綁定不是一回事，這里僅僅就是綁定了一個映射關系，等到后邊事件觸發的時候需要用到該處映射關系
    EventBus.Default.RegisterHandler<EventWithData<Person>, Person_EventHandler>();

    //其它中間件。。
    。。。

    //啟動消費者訂閱監控入口
    app.Consumer(app.ApplicationServices, ServiceCollection);
}

3，啟動消費者訂閱監控，並觸發事件處理程序

/// <summary>
/// 啟動消費者訂閱監控入口
/// </summary>
public static void Consumer(this IApplicationBuilder app, IServiceProvider provider, IServiceCollection services)
{
    //獲取實例對象 
    IMessageQueue queue = provider.GetService<IMessageQueue>();   

    //異步調用，這里接收的消息類型T為：EventWithData<Person>  也可以是其它自定義消息類型,沒有限制
    queue.Consume<EventWithData<Person>>(async message =>
    {
        using (var currentContainer = services.BuildServiceProvider().CreateScope())   //使用系統內置服務容器
        {
            IocManager.Configure(currentContainer.ServiceProvider);                //將容器服務傳遞過去
            
            var e = EventWithData<Person>.New(message.Data);      //這里要傳遞的消息類型必須使用 EventWithData<TData>類型初始化消息
            
            await EventBus.Default.TriggerAsync(e);               //調用觸發事件邏輯(需要提前綁定映射關系和注冊容器)
            
            return true;
        }
    });
}

說明：

IocManager.Configure(currentContainer.ServiceProvider)  實際上就是聲明了一個全局的 IServiceProvider 變量，然后將當前的ServiceProvider賦值給全局變量

await EventBus.Default.TriggerAsync(e)   執行的就是上一節事件總線中的 public async Task TriggerAsync<TEvent>(TEvent e)  方法

到這里 事件源和事件處理程序就一一對應上了。

接下來開始測試一下是否可用，先在發布的應用程序Startup中注冊一下消息隊列容器

services.AddSingleton<IMessageQueue, RabbitMQPublishClient>();

然后發布一條消息，注意：這里消息的類型EventWithData<Person>沒有限制的，你可以不使用EventWithData<Person>類型的消息也行

//發布消息
var message = EventWithData<Person>.New(
    new Person()
    {
        UserId = 1,
        UserName = "張三"
    }
);

messageQueue.Publish(message);

最后測試，訂閱者成功消費了剛剛生產的消息