33 | 集成事件：使用RabbitMQ來實現(xiàn)EventBus

這一節(jié)我們來講解如何通過 CAP 組件和 RabbitMQ 來實現(xiàn) EventBus

要實現(xiàn) EventBus，我們這里借助了 RabbitMQ，它的整個安裝和使用的體驗是非常人性化的，如果是在 Windows 下開發(fā)的話，它可以有 Windows 的 installer，也可以在其它的操作系統(tǒng)下安裝和使用，當(dāng)然它也支持 Docker 的模式，我們可以在以下的地址獲取到安裝包和安裝方法的說明

https://www.rabbitmq.com/download.html

另一個就是在 .NET Core 社區(qū)比較知名的 CAP 框架，這個框架是由我們國人開發(fā)的，它實現(xiàn)了開箱即用的 EventBus 的實現(xiàn)，我們可以通過簡單的配置，就能把 RabbitMQ 集成進(jìn)來，并且實現(xiàn)我們的集成事件的處理

https://github.com/dotnetcore/CAP

我們來看一下 CAP 框架的實現(xiàn)架構(gòu)

它實際上實現(xiàn)了一個叫 OutBox 的設(shè)計模式，就是在我們的每個微服務(wù)，比如說微服務(wù) A 的數(shù)據(jù)庫 A，在這個數(shù)據(jù)庫內(nèi)部它建立了兩張表，一張叫 publish 事件表和一張叫 receiver 事件表，這兩張事件表用來記錄微服務(wù) A 發(fā)出的和微服務(wù) A 收到的事件

當(dāng)我們要發(fā)出事件時，我們會把事件的存儲邏輯與我們的業(yè)務(wù)邏輯的事務(wù)合并，在同一個事務(wù)里提交，也就意味著當(dāng)我們的業(yè)務(wù)邏輯提交成功時，我們的事件表里面的事件是一定存在的，它是與我們的業(yè)務(wù)邏輯的事務(wù)是強(qiáng)綁定的

如果說我們的業(yè)務(wù)邏輯失敗了，事務(wù)回滾了，這條事件是不會出現(xiàn)在我們的事件表里的，這樣子就可以做到說我們要發(fā)送的事件一定是與業(yè)務(wù)邏輯是一致的

接下來由我們組件來負(fù)責(zé)將事件表里的事件全部都發(fā)送到 EventBus，比如說 RabbitMQ 消息隊列里面去，由接收方訂閱

對于訂閱的事件的話，設(shè)計的模式也是同理，當(dāng)我們的應(yīng)用程序在消息隊列獲取到信息的時候，它就會將這些消息持久化到我們的數(shù)據(jù)庫的 Receive 事件表里，這樣我們就可以在本地進(jìn)行事件的處理，失敗重試等操作，這些都是由 CAP 框架完成的，我們僅需要去做簡單的配置，關(guān)注發(fā)布和訂閱的業(yè)務(wù)邏輯即可

我們看一下代碼，剛才有提到 CAP 的架構(gòu)，關(guān)鍵的一點是需要事件的存儲與我們的業(yè)務(wù)邏輯在同一個事務(wù)里，所以說我們在處理事務(wù)的邏輯部分的話，需要嵌入 CAP 的一部分代碼，我們看一下 EFContext 的定義

public EFContext(DbContextOptions options, IMediator mediator, ICapPublisher capBus) : base(options) {_mediator = mediator;_capBus = capBus; }

之前有關(guān)注到有一個叫 ICapPublisher 這個入?yún)?#xff0c;關(guān)鍵的是這一行代碼我們需要關(guān)注一下

_currentTransaction = Database.BeginTransaction(_capBus, autoCommit: false);

這一行代碼的作用是創(chuàng)建事務(wù)，我們可以看到創(chuàng)建事務(wù)的過程中，我們把 ICapPublisher 也傳入給了這個方法的構(gòu)造函數(shù)，實際上這個方法是由 CAP 的組件提供的，它的核心作用就是將我們要發(fā)送的事件與我們的業(yè)務(wù)的存儲都放在同一個事務(wù)內(nèi)部，這樣子我們就可以使得事務(wù)提交時或者回滾時，我們的事件與業(yè)務(wù)邏輯的存取都是一致的

然后我們再來看一下配置的部分，寫在 ServiceCollectionExtensions 下面

public static IServiceCollection AddEventBus(this IServiceCollection services, IConfiguration configuration) {services.AddTransient<ISubscriberService, SubscriberService>();services.AddCap(options =>{options.UseEntityFramework<DomainContext>();options.UseRabbitMQ(options =>{configuration.GetSection("RabbitMQ").Bind(options);});//options.UseDashboard();});return services; }

我們這里定義了一個 AddEventBus，可以看到將我們之前演示的代碼訂閱服務(wù)注入進(jìn)來，然后 Services 最重點的代碼是 AddCap，我們需要告訴 CAP 框架我們是針對 DomainContext 來實現(xiàn)我們的 EventBus，EventBus 與 DomainContext 共享我們的數(shù)據(jù)庫連接，下面一行代碼是指我們要用 RabbitMQ 來作為我們 EventBus 的消息隊列的存儲，這里可以看到使用了一個 Bind 的方法將我們的配置綁定到 RabbitMQ 的 options 上面去

我們可以看一下我們的配置

"RabbitMQ": {"HostName": "localhost","UserName": "admin","Password": "123456","VirtualHost": "geektime","ExchangeName": "geek_queue"}

這里可以看到我們定義了一個 RabbitMQ 的配置，然后這里面會有我們的 host，因為是本地安裝的，所以訪問地址就是 localhost，VirtualHost 是 RabbitMQ 一個比較特殊的設(shè)置，它的作用是將 RabbitMQ 的空間區(qū)分為不同的空間，你可以認(rèn)為這是一個租戶，相同的 VirtualHost，大家都可以認(rèn)為是一個 RabbitMQ 的集群，最下面的 ExchangeName 就是隊列需要訂閱的 Exchange 的名稱，消息的發(fā)布和訂閱都是通過這個 Exchange 來的

然后我們在 Startup 這里添加一行

services.AddEventBus(Configuration);

這樣我們就配置完成了

總結(jié)

以上是生活随笔為你收集整理的.NET Core开发实战（第33课：集成事件：使用RabbitMQ来实现EventBus）--学习笔记（上）...的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。