1. 引言

事件總線解決了微服務(wù)間如何基于集成事件進(jìn)行異步通信的問(wèn)題。然而只有事件總線正常運(yùn)行，微服務(wù)之間基于事件的通信才得以運(yùn)轉(zhuǎn)。 而現(xiàn)實(shí)情況是，總有這樣或那樣的問(wèn)題，導(dǎo)致事件總線不穩(wěn)定或不可用，比如：網(wǎng)絡(luò)中斷，系統(tǒng)斷電等等，這都可能導(dǎo)致微服務(wù)間的不一致性問(wèn)題。 那如何解決事件總線故障導(dǎo)致的不一致問(wèn)題呢？

事件溯源

事件日志挖掘

發(fā)件箱模式

2. 問(wèn)題

既然上面提到了一致性問(wèn)題，那具體的問(wèn)題是什么呢，在什么情況才會(huì)發(fā)生呢？我想我有必要簡(jiǎn)單舉例。上代碼：

var oldPrice = item.Price;

item.Price = product.Price;

_context.CatalogItems.Update(item);

var @event = new ProductPriceChangedIntegrationEvent(item.Id, item.Price, oldPrice);

// Commit changes in original transaction

await _context.SaveChangesAsync();

// Publish integration event to the event bus

// (RabbitMQ or a service bus underneath)

_eventBus.Publish(@event);

當(dāng)產(chǎn)品價(jià)格更改后，代碼將數(shù)據(jù)提交給數(shù)據(jù)庫(kù)，然后發(fā)布 ProductPriceChangedIntegrationEvent 事件。 如果服務(wù)在數(shù)據(jù)庫(kù)更新后崩潰（奔潰發(fā)生在 _context.SaveChangesAsync()代碼執(zhí)行之后，但又發(fā)生在集成事件成功發(fā)布前），就會(huì)導(dǎo)致本地微服務(wù)價(jià)格已成功更新，但集成事件未發(fā)布的問(wèn)題。就會(huì)導(dǎo)致目錄微服務(wù)中定義的價(jià)格和顧客購(gòu)物車(chē)中緩存的價(jià)格不一致。

3. 分析問(wèn)題

以上問(wèn)題的關(guān)鍵在于是如何確保兩個(gè)獨(dú)立的操作的原子性。如果單從單體應(yīng)用的角度來(lái)處理的話，我們完全是可以將他們放到同一個(gè)事務(wù)中去保證。然而在微服務(wù)中，就違背了其高可用的基本要求。因?yàn)橐坏┦录偩€處于癱瘓狀態(tài)，那么整個(gè)目錄微服務(wù)就不可用了。這種強(qiáng)制通過(guò)事務(wù)保證的一致性，就引入了太多的問(wèn)題依賴(lài)。

如果從微服務(wù)的角度來(lái)看，每個(gè)微服務(wù)負(fù)責(zé)各自的業(yè)務(wù)邏輯，對(duì)于目錄微服務(wù)來(lái)說(shuō)，它的關(guān)注點(diǎn)是產(chǎn)品的更新是否成功。至于借助事件總線通過(guò)異步事件實(shí)現(xiàn)微服務(wù)間的通信，并不是其關(guān)注點(diǎn)。這也就是關(guān)注點(diǎn)分離。換句話說(shuō)，產(chǎn)品的更新不應(yīng)該依賴(lài)外部狀態(tài)。在這里，外部狀態(tài)就是事件總線的可用性。

你可能會(huì)說(shuō)了，既然不允許通過(guò)強(qiáng)事務(wù)保證一致性，那么如何解決一致性問(wèn)題呢（好像繞了半天又回到了原點(diǎn)）？

這里就要引入強(qiáng)一致性和最終一致性的概念了。強(qiáng)一致性：也就是事務(wù)一致性，將多個(gè)操作放到單一事務(wù)處理。要么全部成功，要么全部失敗。最終一致性：通過(guò)將某些操作的執(zhí)行延遲到稍后的時(shí)間來(lái)執(zhí)行。若前面的操作執(zhí)行成功，后續(xù)操作將延后執(zhí)行。若前面的操作失敗，后續(xù)的操作就不會(huì)執(zhí)行。

到這里，我們實(shí)際要解決的問(wèn)題就明確了：如何確保事件總線能夠正確進(jìn)行事件轉(zhuǎn)發(fā)？

換句話說(shuō)：事件總線掛了，但是事件消息不能丟失。只要事件消息不丟，后面我們還有機(jī)會(huì)挽救（重新發(fā)布消息）。

如何保證事件消息不丟失呢？當(dāng)然是持久化了。

4. 持久化事件源

eShopOnContainers已經(jīng)考慮了這一點(diǎn)，集成了事件日志用于持久化。我們直接來(lái)看類(lèi)圖：從類(lèi)圖中看其實(shí)現(xiàn)邏輯也很簡(jiǎn)單，主要是定義了一個(gè) IntegrationEventLogEntry實(shí)體、 EventStateEnum事件狀態(tài)枚舉和 IntegrationEventLogContextEF上下文用于事件日志持久化。暴露 IIntegrationEventLogService用于事件狀態(tài)的更新。

其他微服務(wù)通過(guò)在啟動(dòng)類(lèi)中注冊(cè) IntegrationEventLogContext即可完成事件日志的集成。

services.AddDbContext<IntegrationEventLogContext>(options =>

{

? ?options.UseSqlServer(configuration["ConnectionString"],

? ? ? ?sqlServerOptionsAction: sqlOptions =>

? ? ? ?{

? ? ? ? ? ?sqlOptions.MigrationsAssembly(typeof(Startup)

? ? ? ? ? ? ? ?.GetTypeInfo().Assembly.GetName().Name);

? ? ? ? ? ?sqlOptions.EnableRetryOnFailure(maxRetryCount: 10,

? ? ? ? ? ? ? ?maxRetryDelay: TimeSpan.FromSeconds(30),

? ? ? ? ? ? ? ?errorNumbersToAdd: null);

? ? ? ?});

});

使用EF進(jìn)行數(shù)據(jù)庫(kù)遷移后，就會(huì)生成 IntergrationEventLog表。如下圖所示：

5. 如何借助事件日志確保高可用

主要分兩步走：

應(yīng)用程序開(kāi)始本地?cái)?shù)據(jù)庫(kù)事務(wù)，然后更新領(lǐng)域?qū)嶓w狀態(tài)，并將集成事件插入集成事件日志表中，最后提交事務(wù)來(lái)確保領(lǐng)域?qū)嶓w更新和保存事件日志所需的原子性。

發(fā)布事件

第一步毋庸置疑，第二步發(fā)布事件，我們又有多種實(shí)現(xiàn)方式：

在提交事務(wù)后立即發(fā)布集成事件，并將其標(biāo)記為已發(fā)布。當(dāng)微服務(wù)發(fā)生故障時(shí)，可以通過(guò)遍歷存儲(chǔ)的集成事件（未發(fā)布）執(zhí)行補(bǔ)救措施。

將事件日志表用作一種隊(duì)列。使用單獨(dú)的線程或進(jìn)程查詢事件日志表，將事件發(fā)布到事件總 線，然后將事件標(biāo)記為已發(fā)布。

這里很顯然第二種方式更為穩(wěn)妥。而eShopOnContainers出于簡(jiǎn)單考慮，采用了第一種方案，具體代碼如下：

using (var transaction = _catalogContext.Database.BeginTransaction())

{

_catalogContext.CatalogItems.Update(catalogItem);

await _catalogContext.SaveChangesAsync();

// Save to EventLog only if product price changed

if(raiseProductPriceChangedEvent)

await

_integrationEventLogService.SaveEventAsync(priceChangedEvent);

transaction.Commit();

}

// Publish the intergation event through the event bus

_eventBus.Publish(priceChangedEvent);

integrationEventLogService.MarkEventAsPublishedAsync( priceChangedEvent);

至此，eShopOnContainers確保事件總線能夠正確轉(zhuǎn)發(fā)消息的解決方案闡述完畢。你可能會(huì)問(wèn)，這對(duì)應(yīng)的是引言中的哪一種方案？都不是，你可以看作其是基于事件日志的簡(jiǎn)化版的事件溯源。

6. 僅此而已？

通過(guò)持久化事件日志來(lái)避免事件發(fā)布失敗導(dǎo)致的一致性問(wèn)題，是一種有效措施。然而消息從發(fā)送到接收再到正常消費(fèi)的過(guò)程中，每一個(gè)環(huán)節(jié)都可能故障，所以僅僅在消息發(fā)送端使用事件日志只是確保最終一致性的一小步。還有很多問(wèn)題有待完善：

消息發(fā)送成功了，但未被成功接收

消息發(fā)送且成功接收，但未被正確消費(fèi)

消息重復(fù)發(fā)送，導(dǎo)致多次消費(fèi)問(wèn)題

消息被多個(gè)微服務(wù)訂閱，如何確保每個(gè)微服務(wù)都成功接收并消費(fèi)

等等

而這些問(wèn)題就留給大家思考吧。

總結(jié)

以上是生活随笔為你收集整理的eShopOnContainers 知多少[6]：持久化事件日志的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

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