.NET Core微服務之基於EasyNetQ使用RabbitMQ訊息佇列

阿新 • • 發佈：2019-01-16

一、訊息佇列與RabbitMQ

1.1 訊息佇列

　　“訊息”是在兩臺計算機間傳送的資料單位。訊息可以非常簡單，例如只包含文字字串；也可以更復雜，可能包含嵌入物件。訊息被髮送到佇列中，“訊息佇列”是在訊息的傳輸過程中儲存訊息的容器。

　　訊息佇列（Message Queue），是分散式系統中重要的元件，其通用的使用場景可以簡單地描述為：

當不需要立即獲得結果，但是併發量又需要進行控制的時候，差不多就是需要使用訊息佇列的時候。　　

　　訊息佇列主要解決了應用耦合、非同步處理、流量削鋒等問題。當前使用較多的訊息佇列有RabbitMQ、RocketMQ、ActiveMQ、Kafka、ZeroMQ、MetaMq等，而部分資料庫如Redis、Mysql以及phxsql也可實現訊息佇列的功能。更多詳細內容請參考：《

訊息佇列及其應用場景介紹》

　　我也在前幾年寫過一篇基於Redis做訊息佇列的文章，對訊息佇列的一個應用場景做了介紹，沒有了解過的童鞋可以看看。

1.2 RabbitMQ

　　RabbitMQ是一款基於AMQP（高階訊息佇列協議），由Erlang開發的開源訊息佇列元件。是一款優秀的訊息佇列元件，他由兩部分組成：服務端和客戶端，客戶端支援多種語言的驅動，如：.Net、JAVA、Erlang等。

　　網上有很多效能比較的文章，例如在1百萬條1k的訊息下，每秒種的收發情況如下圖所示：

　　效能比較

　　這裡不過多介紹RabbitMQ，有關RabbitMQ的一些需要了解的概念你可以通過下面的文章瞭解：

　　如果你想了解RabbitMQ與Kafka的對比，可以閱讀這篇文章：《開源軟體成熟度評測報告-分散式訊息中介軟體》

　　而EasyNetQ呢，它是一款基於RabbitMQ.Client封裝的API庫，正如其名，使用起來比較Easy，它把原RabbitMQ.Client中的很多操作都進行了再次封裝，讓開發人員減少了很多工作量。

二、RabbitMQ的安裝

2.1 Linux下的安裝

　　這裡不演示如何在Linux下安裝，但推薦生產環境使用Linux，下面是一些參考資料：

2.2 Windows下的安裝

　　開發環境下，我一般使用Windows Server虛擬機器，所以這裡說明下如何在Windows下安裝：

　　（1）下載Erlang和RabbitMQ (這裡我選則的並非最新版本，而是etp20.3和rabbitmq3.7.5)

　　（2）首先安裝Erlang，然後新增環境變數（如果添加了，則skip這一步）並加到PATH中

　　（3）其次安裝RabbitMQ，一路Next，安裝完成後也為其新增環境變數並新增到PATH中

　　（4）檢查是否安裝成功：rabbitmqctl status

　　這裡我碰到了如下的錯誤：

　　解決方法：

　　最終狀態：

　　檢查Windows服務，發現已經自動註冊了一個服務：

　　（5）啟用Web管理外掛，然後檢查是否可見（http://127.0.0.1:15672）

2.3 一些必要的配置

　　（1）使用預設賬號:guest/guest登入進去，新增一個新使用者（Administrator許可權），並設定其Permission

　　（2）新增新的虛擬機器（預設為/，這裡我新增一個名為EDCVHOST的虛擬機器）

　　（3）繫結新新增的使用者到新的虛擬機器上，接下來在我們的程式中就主要使用admin這個使用者和EDCVHOST這個虛擬機器

　　*.當然，為了安全考慮，你也可以把guest使用者remove掉

三、Quick Start:第一個訊息佇列

3.1 專案準備

　　這裡為了快速的演示如何使用EasyNetQ，我們來一個QuickStart，準備三個專案：兩個Console程式和一個Class Library。

　　其中，對Publisher和Subscriber專案安裝EasyNetQ：

NuGet>Install-Package EasyNetQ　　

　　針對Messages類庫，新增一個class如下：

    public class TextMessage
    {
        public string Text { get; set; }
    }

3.2 我是Publisher

　　新增以下程式碼：

    public class Program
    {
        public static void Main(string[] args)
        {
            var connStr = "host=192.168.80.71;virtualHost=EDCVHOST;username=admin;password=edison";

            using (var bus = RabbitHutch.CreateBus(connStr))
            {
                var input = "";
                Console.WriteLine("Please enter a message. 'Quit' to quit.");
                while ((input = Console.ReadLine()) != "Quit")
                {
                    bus.Publish(new TextMessage
                    {
                        Text = input
                    });
                }
            }
        }
    }

　　可以看到，我們在其中使用EasyNetQ高度封裝的介面建立了一個IBus介面的例項，通過這個IBus例項我們可以通過一個超級Easy的Publish介面進行釋出訊息。這裡主要是讀取使用者在控制檯中輸入的訊息字串進行傳送。實際中，傳送的一般都是一個或多個複雜的實體物件。

3.3 我是Subscriber

　　新增如下所示程式碼：

    public class Program
    {
        public static void Main(string[] args)
        {
            var connStr = "host=192.168.80.71;virtualHost=EDCVHOST;username=admin;password=edison";

            using (var bus = RabbitHutch.CreateBus(connStr))
            {
                bus.Subscribe<TextMessage>("my_test_subscriptionid", HandleTextMessage);

                Console.WriteLine("Listening for messages. Hit <return> to quit.");
                Console.ReadLine();
            }
        }

        public static void HandleTextMessage(TextMessage textMessage)
        {
            Console.ForegroundColor = ConsoleColor.Red;
            Console.WriteLine("Got message: {0}", textMessage.Text);
            Console.ResetColor();
        }
    }

　　這裡主要是通過IBus例項去訂閱訊息（這裡是除非使用者關閉程式否則一直處於監聽狀態），當釋出者釋出了指定型別的訊息之後，這裡就把它打印出來（紅色字型顯示）。

3.4 簡單測試

　　通過控制檯資訊檢視結果：

　　通過RabbitMQ管理介面檢視：

　　（1）通過Connections Tab可以發現我們的兩個客戶端都在Running中

　　（2）通過Queues Tab檢視目前已有的佇列=>可以看到目前我們只註冊了一個佇列

四、在ASP.NET Core中的使用

4.1 案例結構與說明

　　這裡假設有這樣一個場景，客戶通過瀏覽器提交了一個保單，這個保單中包含一些客戶資訊，ClientService將這些資訊處理後傳送一個訊息到RabbitMQ中，NoticeService和ZAPEngineService訂閱了這個訊息。NoticeService會將客戶資訊取出來並獲取一些更多資訊為客戶傳送Email，而ZAPEngineService則會根據客戶的一些關鍵資訊（比如：年齡，是否吸菸，學歷，年收入等等）去資料庫讀取一些規則來生成一份Question List並存入資料庫。

4.2 專案準備工作

　　建立上面提到的這幾個專案，這裡我選擇ASP.NET Core WebAPI型別。

　　分別為這幾個專案通過NuGet安裝EasyNetQ元件，並且通過以下程式碼注入統一的IBus例項物件：

    public IServiceProvider ConfigureServices(IServiceCollection services)
    {
      // IoC - EventBus
      services.AddSingleton(RabbitHutch.CreateBus(Configuration["MQ:Dev"]));
      ......
    }

　　這裡我將連線字串寫到了配置檔案中，請參考上面的QuickStart中的內容。

　　下面是這個demo用到的一個訊息物件實體：通過標籤宣告佇列名稱。

    [Queue("Qka.Client", ExchangeName = "Qka.Client")]
    public class ClientMessage
    {
        public int ClientId { get; set; }
        public string ClientName { get; set; }
        public string Sex { get; set; }
        public int Age { get; set; }
        // N: Non-Smoker, S: Smoker
        public string SmokerCode { get; set; }
        // Bachelor, Master, Doctor
        public string Education { get; set; }
        public decimal YearIncome { get; set; }
    }

　　這裡為了快速的在專案中使用Subscriber，新增一個擴充套件方法，它會從注入的服務中取出IBus例項物件，並自動幫我們進行Subscriber（那些實現了IConsume介面的類）的註冊。具體用法見後面的介紹。

    public static class AppBuilderExtension
    {
        public static IApplicationBuilder UseSubscribe(this IApplicationBuilder appBuilder, string subscriptionIdPrefix, Assembly assembly)
        {
            var services = appBuilder.ApplicationServices.CreateScope().ServiceProvider;

            var lifeTime = services.GetService<IApplicationLifetime>();
            var bus = services.GetService<IBus>();
            lifeTime.ApplicationStarted.Register(() =>
            {
                var subscriber = new AutoSubscriber(bus, subscriptionIdPrefix);
                subscriber.Subscribe(assembly);
                subscriber.SubscribeAsync(assembly);
            });

            lifeTime.ApplicationStopped.Register(() => bus.Dispose());

            return appBuilder;
        }
    }

4.3 Publisher：ClientService

　　ClientService作為釋出者，這裡假設我們在API中處理完業務程式碼後，將message釋出給RabbitMQ：

    [Produces("application/json")]
    [Route("api/Client")]
    public class ClientController : Controller
    {
        private readonly IClientService clientService;
        private readonly IBus bus;

        public ClientController(IClientService _clientService, IBus _bus)
        {
            clientService = _clientService;
            bus = _bus;
        }

        ......

        [HttpPost]
        public async Task<string> Post([FromBody]ClientDTO clientDto)
        {
            // Business Logic here...
            // eg.Add new client to your service databases via EF
            // Sample Publish
            ClientMessage message = new ClientMessage
            {
                ClientId = clientDto.Id.Value,
                ClientName = clientDto.Name,
                Sex = clientDto.Sex,
                Age = 29,
                SmokerCode = "N",
                Education = "Master",
                YearIncome = 100000
            };
            await bus.PublishAsync(message);

            return "Add Client Success! You will receive some letter later.";
        }
    }

　　當然，你可以使用同步方法：bus.Publish(message);

4.4 Subscriber: NoticeService & ZAPEngineService

　　（1）NoticeService：新增一個實現IConsume介面的Consumer類

    public class ClientMessageConsumer: IConsumeAsync<ClientMessage>
    {
        [AutoSubscriberConsumer(SubscriptionId = "ClientMessageService.Notice")]
        public Task ConsumeAsync(ClientMessage message)
        {
            // Your business logic code here
            // eg.Build one email to client via SMTP service
            // Sample console code
            System.Console.ForegroundColor = System.ConsoleColor.Red;
            System.Console.WriteLine("Consume one message from RabbitMQ : {0}, I will send one email to client.", message.ClientName);
            System.Console.ResetColor();

            return Task.CompletedTask;
        }
    }

　　這裡為了演示效果，增加了一些輸出資訊的程式碼，下面的ZAPEngineService也是一樣，不再贅述。

　　（2）ZAPEngineService：新增一個實現IConsume介面的Consumer類

    public class ClientMessageConsumer : IConsumeAsync<ClientMessage>
    {
        [AutoSubscriberConsumer(SubscriptionId = "ClientMessageService.ZapQuestion")]
        public Task ConsumeAsync(ClientMessage message)
        {
            // Your business logic code here
            // eg.Generate one ZAP question records into database and send to client
            // Sample console code
            System.Console.ForegroundColor = System.ConsoleColor.Red;
            System.Console.WriteLine("Consume one message from RabbitMQ : {0}, I will generate one ZAP question list to client", message.ClientName);
            System.Console.ResetColor();

            return Task.CompletedTask;
        }
    }

　　注意兩個Consumer的SubscriptionId不能一樣，否則無法接受到訊息。

　　（3）為兩個Consumer使用擴充套件方法：UseSubscribe

    public void Configure(IApplicationBuilder app, IHostingEnvironment env, IApplicationLifetime lifetime)
    {
        ......

        // easyNetQ
        app.UseSubscribe("ClientMessageService", Assembly.GetExecutingAssembly());
    }

4.5 簡單測試

　　（1）藉助Postman向ClientService發起Post請求

　　（2）檢視NoticeService的日誌資訊

　　（3）檢視ZAPEngineService的日誌資訊

　　（4）檢視RabbitMQ的管理控制檯：

五、小結

　　本篇超級簡單地介紹了一下訊息佇列與RabbitMQ，通過使用EasyNetQ這個基於RabbitMQ.Client的客戶端做了一個QuickStart演示了在.NET Core環境下如何進行訊息的釋出與訂閱，並通過一個微服務的小案例演示瞭如何在ASP.NET Core環境下如何基於EasyNetQ完成訊息的釋出與訂閱，看起來就像一個類似於簡單的事件匯流排。當然，本篇的內容都十分基礎，如果要應用好RabbitMQ，還得把那些基礎概念（如：Channel，Exchange等）弄清楚，然後去理解一下事件匯流排的概念，實際中還得考慮資料一致性等等，路途漫漫，繼續加油吧！

示例程式碼

　　Click Here => 點我下載

參考資料

作者：周旭龍

本文版權歸作者和部落格園共有，歡迎轉載，但未經作者同意必須保留此段宣告，且在文章頁面明顯位置給出原文連結。