🥇具有在 Kafka 上使用非同步 API 開發退款工具服務的經驗

是什麼會迫使像 Lamoda 這樣擁有簡化流程和數十項相互關聯的服務的大公司顯著改變其方法？動機可能完全不同：從立法到所有程式設計師固有的實驗願望。

但這並不意味著您不能指望額外的好處。 Sergey Zaika 將告訴您，如果您在 Kafka 上實現事件驅動的 API，您到底能贏得什麼（弗里沃德）。肯定還會談論大人物和有趣的發現——實驗離不開它們。

免責聲明：本文基於 Sergey 於 2018 年 XNUMX 月在 HighLoad++ 上舉辦的一次聚會的資料。 Lamoda 與 Kafka 合作的現場體驗吸引了聽眾，其程度不亞於日程上的其他報導。我們認為這是一個很好的例子，說明您可以而且應該始終找到志同道合的人，HighLoad++ 的組織者將繼續努力營造有利於這一點的氛圍。

關於流程

Lamoda 是一個大型電子商務平台，擁有自己的聯絡中心、送貨服務（以及許多附屬機構）、照相館、龐大的倉庫，所有這些都運作在自己的軟體上。支付方式有數十種，B2B 合作夥伴可能會使用其中部分或全部服務，並希望了解其產品的最新資訊。此外，Lamoda 在俄羅斯聯邦以外的三個國家開展業務，那裡的一切都有點不同。總共可能有一百多種配置新訂單的方法，必須以自己的方式處理。所有這一切都需要數十種服務的幫助，這些服務有時會以不明顯的方式進行溝通。還有一個中央系統，其主要職責是訂單狀態。我們叫她 BOB，我和她一起工作。

具有事件驅動 API 的退款工具

事件驅動這個詞已經很陳腔濫調了；稍後我們將更詳細地定義它的意義。我將從我們決定在 Kafka 中嘗試事件驅動的 API 方法的背景開始。

在任何商店裡，除了顧客付款的訂單外，有時也會因為產品不適合顧客而被要求退貨。這是一個相對較短的過程：如有必要，我們會澄清資訊並轉移。

但由於立法的變化，退貨變得更加複雜，我們必須為其實現單獨的微服務。

我們的動機：

法FZ-54 - 簡而言之，法律要求在幾分鐘的相當短的 SLA 內向稅務局報告每筆貨幣交易，無論是申報表還是收據。我們作為一家電子商務公司，開展了相當多的業務。從技術上講，這意味著新的責任（因此也意味著新的服務）和所有相關係統的改進。
鮑伯分裂 是公司內部項目，旨在將BOB從大量非核心職責中解放出來，降低其整體複雜性。

該圖顯示了主要的 Lamoda 系統。現在大部分都比較 圍繞著不斷縮小的整體架構的 5-10 個微服務群。它們正在緩慢增長，但我們正在努力使它們變小，因為部署中間選擇的片段是可怕的 - 我們不能讓它掉落。我們被迫保留所有交易所（箭頭），並考慮到其中任何一個交易所可能無法使用的事實。

BOB也有相當多的交易所：支付系統、配送系統、通知系統等。

從技術上講，BOB 是：

~150k 行程式碼 + ~100k 行測試；
php7.2 + Zend 1 和 Symfony 組件 3；
>100 個 API 和約 50 個出站整合；
4個國家有自己的商業邏輯。

部署 BOB 既昂貴又痛苦，其程式碼量和解決的問題之多，沒有人能把它們全部記在腦子裡。一般來說，簡化它的原因有很多。

退貨流程

最初，該過程涉及兩個系統：BOB 和支付。現在又出現了兩個：

財政化服務，將解決財政化以及與外部服務溝通的問題。
退款工具，僅包含新的交易所，以免導致 BOB 膨脹。

現在該過程如下所示：

BOB 收到退款要求。
BOB 談論這個退款工具。
退款工具告訴付款：“退還錢。”
付款返還錢。
退款工具和 BOB 相互同步狀態，因為目前他們都需要它。我們還沒有準備好完全切換到退款工具，因為 BOB 有使用者介面、會計報告以及一般情況下無法輕鬆傳輸的大量資料。你必須坐在兩張椅子上。
財政化的要求消失了。

於是，我們在 Kafka 上做了一個事件總線——event-bus，一切都從它開始。萬歲，現在我們遇到了單點故障（諷刺）。

優點和缺點都非常明顯。我們製造了一輛公共汽車，這意味著現在所有的服務都依賴它。這簡化了設計，但會在系統中引入單點故障。 Kafka 將崩潰，進程將停止。

什麼是事件驅動 API

Martin Fowler 的報告對此問題給了一個很好的答案（GOTO 2017） “事件驅動架構的多種意義”.

簡單來說我們做了什麼：

透過以下方式封裝所有非同步交換 事件儲存。我們不是透過網路通知每個感興趣的消費者有關狀態更改的信息，而是將有關狀態更改的事件寫入集中存儲，並對主題感興趣的消費者讀取從那裡出現的所有內容。
本例中的事件是通知 (通知）某處發生了變化。例如，訂單狀態會發生變化。如果消費者對通知中未包含的伴隨狀態變化的某些數據感興趣，則可以自行了解其狀態。
最大的選擇是成熟的事件溯源， 狀態轉移，其中事件包含處理所需的所有資訊：它來自哪裡、它進入什麼狀態、資料到底如何更改等。唯一的問題是可行性以及您可以儲存的資訊量。

作為退款工具啟動的一部分，我們使用了第三種選項。這簡化了事件處理，因為不需要提取詳細信息，而且它消除了每個新事件都會產生來自消費者的大量澄清獲取請求的情況。

退款工具服務 未載入，所以卡夫卡的存在更多的是筆的品味而不是必需品。我不認為如果退款服務成為一個高負荷的項目，生意會很高興。

非同步交換原樣

對於非同步交換，PHP部門通常使用RabbitMQ。我們收集請求的數據，將其放入隊列中，同一服務的用戶讀取它並發送它（或不發送它）。對於 API 本身，Lamoda 積極使用 Swagger。我們設計一個 API，用 Swagger 描述它，並產生客戶端和伺服器程式碼。我們也使用了稍微增強的 JSON RPC 2.0。

在某些地方使用 ESB 總線，有些地方使用 activeMQ，但一般來說， RabbitMQ - 標準.

非同步交換即將到來

當透過事件總線設計交換時，可以進行類比。我們同樣透過事件結構描述來描述未來的資料交換。 yaml 格式，我們必須自己進行程式碼生成，生成器根據規範建立 DTO，並指導客戶端和伺服器使用它們。一代人進入兩種語言- golang 和 php。這有助於保持庫的一致性。該生成器是用 golang 編寫的，這就是它被稱為 gogi 的原因。

Kafka 上的事件溯源是一件典型的事。 Kafka Confluence主力企業版有一個解決方案，有納卡迪，來自我們域兄弟 Zalando 的解決方案。我們的 開始使用普通 Kafka 的動機 - 這意味著在我們最終決定是否在所有地方使用該解決方案之前，該解決方案是免費的，同時也為我們自己留有迴旋和改進的空間：我們希望為我們的解決方案提供支持 JSON RPC 2.0，兩種語言的生成器，讓我們看看還有什麼。

諷刺的是，即使在這樣一個令人愉快的案例中，當有一個大致相似的業務，Zalando，它提出了大致相似的解決方案時，我們卻無法有效地使用它。

啟動時的架構模式如下：我們直接從 Kafka 讀取，但僅透過 events-bus 寫入。 Kafka 中有很多內容可供閱讀：經紀人、平衡器，而且它或多或少已經為水平擴展做好了準備，我想保留這一點。我們希望透過一個網關（又稱事件匯流排）完成錄製，原因如下。

活動巴士

或活動巴士。這只是一個無狀態的 http 網關，它承擔幾個重要的角色：

生產驗證 — 我們檢查活動是否符合我們的規格。
事件主控系統，也就是說，這是公司中主要且唯一的系統，它回答了哪些事件和哪些結構被認為是有效的問題。驗證僅涉及資料類型和枚舉來嚴格指定內容。
哈希函數 對於分片 - Kafka 訊息結構是鍵值對，並使用鍵的雜湊值來計算將其放在哪裡。

為什麼

我們在一家流程精簡的大公司工作。為什麼要改變什麼？ 這是一個實驗，我們預計會獲得一些好處。

1:n+1交換（一對多）

Kafka 讓將新消費者連接到 API 變得非常容易。

假設您有一個目錄，需要同時在多個系統（以及一些新系統）中保持最新。之前，我們發明了一個實作 set-API 的包，並且主系統被告知消費者地址。現在主系統發送該主題的更新，每個有興趣的人都可以閱讀。一個新系統出現了 - 我們為該主題註冊了它。是的，也可以捆綁，但更簡單。

以refund-tool為例，它是BOB的一部分，我們可以方便地透過Kafka來保持它們的同步。 Payment 說錢已退回：BOB、RT 發現了此事，更改了狀態，Fiscalization Service 發現了此事並簽發了支票。

我們計劃創建一個統一的通知服務，通知客戶有關其訂單/退貨的新聞。現在這個責任分散在系統之間。對我們來說，教導通知服務從 Kafka 捕獲相關資訊並對其做出回應（並在其他系統中停用這些通知）就足夠了。不需要新的直接交換。

數據驅動

系統之間的訊息變得透明——無論您擁有什麼樣的“血腥企業”，也無論您的積壓工作有多麼豐富。 Lamoda 擁有一個數據分析部門，負責從系統收集數據並將其轉化為可重複使用的形式，用於業務和智慧系統。 Kafka 允許您快速向他們提供大量資料並保持資訊流最新。

複製日誌

訊息在被讀取後不會消失，就像在 RabbitMQ 中一樣。當事件包含足夠的資訊進行處理時，我們就有了該物件最近更改的歷史記錄，如果需要，還可以套用這些變更。

複製日誌的儲存週期取決於該主題的寫入強度；Kafka可讓您靈活設定儲存時間和資料量的限制。對於密集型主題，重要的是所有消費者都有時間在資訊消失之前閱讀該訊息，即使是在短期無法操作的情況下。通常可以儲存數據 天的單位，這足以提供支持。

接下來稍微複述一下文檔，給那些不熟悉Kafka的人（圖片也來自文檔）

AMQP 有佇列：我們將訊息寫入消費者的佇列。通常，一個佇列由具有相同業務邏輯的一個系統處理。如果您需要通知多個系統，您可以教導應用程式寫入多個佇列或使用扇出機製配置交換，扇出機制會複製它們本身。

Kafka也有類似的抽象主題，您可以在其中寫入訊息，但它們在閱讀後不會消失。預設情況下，當您連接到 Kafka 時，您會收到所有訊息，並可以選擇儲存到上次中斷的位置。也就是說，您按順序閱讀，您可能不會將訊息標記為已讀，而是將 id，然後您可以從中繼續閱讀。你確定的Id叫做offset，機制就是commit offset。

因此，可以實現不同的邏輯。例如，我們在不同國家的 4 個實例中擁有 BOB - Lamoda 在俄羅斯、哈薩克、烏克蘭、白俄羅斯。由於它們是單獨部署的，因此它們的配置和業務邏輯略有不同。我們在訊息中指出它指的是哪個國家。每個國家的每個 BOB 消費者都使用不同的 groupId 來讀取，如果該訊息不適用於他們，他們就會跳過它，即立即提交偏移量+1。如果我們的支付服務讀取相同主題，那麼它會使用單獨的群組來讀取，因此偏移量不會相交。

活動要求：

資料完整性。 我希望該事件有足夠的數據以便可以對其進行處理。

廉正我們委託事件總線驗證事件是否一致並且它可以處理它。
順序很重要。 在回歸的情況下，我們被迫與歷史合作。對於通知，順序並不重要，如果它們是同類通知，則無論哪個訂單先到達，電子郵件都將是相同的。在退款的情況下，有一個明確的流程；如果我們更改訂單，則會出現異常，退款將不會被創建或處理 - 我們最終會處於不同的狀態。
一致性。 我們有一個商店，現在我們創建事件而不是 API。我們需要一種方法來快速、廉價地將有關新事件和現有事件變更的資訊傳輸到我們的服務。這是透過單獨的 git 儲存庫和程式碼產生器中的通用規格來實現的。因此，不同服務中的客戶端和伺服器是協調的。

卡夫卡在拉莫達

我們安裝了三個 Kafka：

紀錄;
研發；
事件總線。

今天我們只討論最後一點。在 events-bus，我們沒有非常大的安裝 - 3 個代理（伺服器）和只有 27 個主題。通常，一個主題就是一個過程。但這是一個微妙的點，我們現在就談談它。

上面是rps圖。退款過程以綠松石線標記（是的，X軸上的那條），粉紅色線是內容更新過程。

Lamoda目錄包含數百萬種產品，資料一直在更新。有些系列已經過時了，新的系列被發布來取代它們，並且新的型號不斷出現在目錄中。我們試圖預測客戶明天會感興趣什麼，因此我們不斷購買新東西，拍攝它們並更新展示櫃。

粉紅色峰值是產品更新，即產品的變化。看得出來，小伙子們拍照、拍照、然後再拍照！ — 載入了一組事件。

Lamoda 活動用例

我們使用建置的架構進行以下操作：

退貨狀態追蹤：來自所有相關係統的號召性用語和狀態追蹤。付款、狀態、財務化、通知。在這裡，我們測試了該方法，製作了工具，收集了所有錯誤，編寫了文件並告訴我們的同事如何使用它。
更新產品卡： 配置、元資料、特徵。一個系統讀取（顯示），多個系統寫入。
電子郵件、推播和簡訊：訂單已取貨、訂單已到、已接受退貨等等，有很多。
庫存、倉庫更新 — 物品的量化更新，只是數字：到達倉庫、退貨。所有與預訂貨物相關的系統都必須使用最新數據運作。目前，庫存更新系統相當複雜；Kafka將簡化它。
數據分析 （研發部門）、機器學習工具、分析、統計。我們希望資訊透明——Kafka 非常適合這一點。

現在更有趣的部分是關於過去六個月中發生的重大變化和有趣的發現。

設計問題

假設我們想做一件新事——例如，將整個交付流程轉移到 Kafka。現在，該流程的一部分已在 BOB 的訂單處理中實現。將訂單轉移到送貨服務、移動到中間倉庫等背後都有一個狀態模型。有一個完整的整體，甚至兩個，加上一堆專用於交付的 API。他們對交付了解更多。

這些看似相似的區域，但 BOB 中的訂單處理和運輸系統具有不同的狀態。例如，某些快遞服務不會發送中間狀態，而僅發送最終狀態：「已送達」或「遺失」。相反，其他人則詳細報告了貨物的流動情況。每個人都有自己的驗證規則：對某些人來說，電子郵件是有效的，這意味著它將被處理；對某些人來說，電子郵件是有效的，這意味著它將被處理；對於其他人來說是無效的，但是訂單仍然會被處理，因為有聯絡電話號碼，有人會說這樣的訂單根本不會被處理。

資料流

就 Kafka 而言，出現了組織資料流的問題。這項任務涉及根據幾個要點選擇策略；讓我們逐一討論一下。

在一個主題還是在不同的主題中？

我們有一個事件規範。在BOB中我們寫某某訂單需要出貨，並註明：訂單號碼、組成、一些SKU和條碼等。當貨物到達倉庫時，交貨將能夠接收狀態、時間戳記和所需的一切。但是我們希望在 BOB 中接收此資料的更新。我們有一個從交付接收資料的逆過程。這是同一個事件嗎？或者這是一個值得擁有自己主題的單獨交易所？

最有可能的是，它們將非常相似，並且創建主題的誘惑並非沒有根據，因為一個單獨的主題意味著單獨的消費者、單獨的配置、所有這些的單獨生成。但這不是事實。

新領域還是新事件？

但如果使用相同的事件，則會出現另一個問題。例如，並非所有交付系統都可以產生 BOB 可以產生的那種 DTO。我們向他們發送 id，但他們沒有保存它，因為他們不需要它，並且從啟動事件總線流程的角度來看，該欄位是必需的。

如果我們為事件匯流排引入一條規則，要求該欄位是必需的，那麼我們就被迫在 BOB 或啟動事件處理程序中設定額外的驗證規則。驗證開始在整個服務中傳播——這不是很方便。

另一個問題是增量開發的誘惑。我們被告知需要在該事件中添加一些內容，如果我們考慮一下，也許它應該是一個單獨的事件。但在我們的方案中，單獨的事件是單獨的主題。一個單獨的主題是我上面描述的整個過程。開發人員很想簡單地在 JSON 模式添加另一個欄位並重新產生它。

在退款的情況下，我們在半年內到達了活動現場。我們有一個稱為退款更新的元事件，它有一個類型欄位描述此更新的實際內容。因此，我們與驗證器進行了「美妙」的切換，驗證器告訴我們如何使用這種類型來驗證此事件。

事件版本控制

要驗證 Kafka 中的消息，您可以使用 Avro公司，但有必要立即躺在上面並使用 Confluence。在我們的例子中，我們必須小心版本控制。由於模型已“離開”，因此並非總是可以從複製日誌中重新讀取訊息。基本上，結果是建立版本以使模型向後相容：例如，使欄位暫時可選。如果差異太大，我們會開始撰寫新主題，並在客戶讀完舊主題後轉移客戶。

保證分區的讀取順序

Kafka 內部的主題分為多個分區。當我們設計實體和交易所時，這並不是很重要，但在決定如何使用和擴展它時，這一點很重要。

通常情況下，您在 Kafka 中編寫一個主題。預設情況下，使用一個分區，本主題中的所有訊息都會轉到該分區。因此，消費者會依序讀取這些訊息。假設現在我們需要擴展系統，以便兩個不同的消費者可以讀取訊息。例如，如果您正在發送短信，那麼您可以告訴 Kafka 進行額外的分區，Kafka 將開始將訊息分成兩部分 - 一半在這裡，一半在這裡。

卡夫卡如何劃分它們？每個訊息都有一個正文（我們在其中儲存 JSON）和一個密鑰。您可以將雜湊函數附加到該鍵，該函數將確定訊息將進入哪個分區。

在我們的退款案例中，這一點很重要，如果我們採用兩個分區，那麼並行消費者有可能會在第一個事件之前處理第二個事件，並且會出現麻煩。雜湊函數確保具有相同金鑰的訊息最終位於同一分區中。

事件與命令

這是我們遇到的另一個問題。 Event 是某個事件：我們說某處發生了某事（something_happened），例如，某件商品被取消或發生退款。如果有人監聽這些事件，那麼根據“項目取消”，將建立退款實體，並且“退款發生”將寫入設定中的某處。

但通常，當您設計事件時，您不想徒勞地編寫它們 - 您依賴於有人會閱讀它們的事實。人們很容易不寫「something_happened」（item_canceled、refund_refunded），而是寫「something_should_be_done」。例如，物品已準備好退貨。

一方面，它顯示瞭如何使用該事件。另一方面，它聽起來不太像一個正常的事件名稱。此外，距離 do_something 指令並不遠。但您不能保證有人閱讀此事件；如果你讀了，那麼你就讀成功了；如果你成功地閱讀了它，那麼你就做了某件事，某件事成功了。一旦事件變成 do_something，回饋就變得必要，這就是一個問題。

在 RabbitMQ 的非同步交換中，當您讀取訊息時，請前往 http，您會得到一個回應 - 至少訊息已收到。當您向 Kafka 寫入資料時，您會向 Kafka 寫入一條訊息，但您不知道它是如何處理的。

因此，在我們的例子中，我們必須引入一個回應事件並設定監視，以便如果發送了這麼多事件，那麼在某個時間之後，應該有相同數量的回應事件到達。如果這沒有發生，那麼似乎出了什麼問題。例如，如果我們發送「item_ready_to_refund」事件，我們預計將建立退款，資金將退還給客戶，並且「money_refunded」事件將發送給我們。但這是不確定的，所以需要監控。

細微之處

有一個相當明顯的問題：如果你按順序讀取一個主題，並且有一些不好的消息，那麼消費者就會下降，你就不會再繼續下去了。你需要 停止所有消費者，進一步提交偏移量以繼續閱讀。

我們知道它，我們指望它，但它還是發生了。發生這種情況是因為從事件匯流排的角度來看該事件是有效的，從應用程式驗證器的角度來看事件是有效的，但從PostgreSQL 的角度來看它是無效的，因為在我們的系統中MySQL對於 UNSIGNED INT，系統只有 INT 的 PostgreSQL。他的尺寸有點小，身分證不適合。 Symfony 例外地去世了。當然，我們捕獲了異常，因為我們依賴它，並且要提交此偏移量，但在此之前我們想要增加問題計數器，因為訊息處理不成功。這個專案中的計數器也在資料庫中，而Symfony已經關閉了與資料庫的通信，第二個異常殺死了整個進程，沒有機會提交偏移量。

該服務擱置了一段時間——幸運的是，對於 Kafka，情況並沒有那麼糟糕，因為消息仍然存在。待工作恢復後，即可閱讀完畢。很舒服。

Kafka 能夠透過工具設定任意偏移量。但要做到這一點，您需要停止所有消費者 - 在我們的例子中，準備一個單獨的版本，其中不會有消費者，重新部署。然後在 Kafka 中，您可以透過工具移動偏移量，訊息就會通過。

另一個細微差別—— 複製日誌與 rdkafka.so - 與我們專案的具體情況有關。我們使用 PHP，在 PHP 中，通常所有函式庫都透過 rdkafka.so 儲存庫與 Kafka 通信，然後有某種包裝器。也許這些都是我們個人的困難，但事實證明，光是重讀我們已經讀過的一段內容並不那麼容易。一般來說，存在軟體問題。

回到使用分區的細節，它在文檔中寫得正確 消費者 >= 主題分區。但我發現這一點的時間比我希望的要晚得多。如果您想要擴展並擁有兩個消費者，則至少需要兩個分割區。也就是說，如果您有一個分區，其中累積了 20 萬個訊息，並且您建立了一個新分區，則訊息數量不會很快平衡。因此，為了擁有兩個並行的消費者，你需要處理分區。

監控

我認為我們監控的方式會更清楚現有方法有哪些問題。

例如，我們計算資料庫中有多少產品最近更改了狀態，相應地，根據這些變更應該發生事件，並將此數字傳送到我們的監控系統。然後從 Kafka 中我們得到第二個數字，即實際記錄了多少事件。顯然，這兩個數字之間的差異應該始終為零。

另外，還需要監控生產者在做什麼，events-bus是否收到訊息，以及消費者在做什麼。例如，在下面的圖表中，退款工具表現良好，但 BOB 顯然存在一些問題（藍色高峰）。

我已經提到過消費者群體的滯後。粗略地說，這是未讀訊息的數量。一般來說，我們的消費者工作速度很快，所以滯後通常是0，但有時也會出現短暫的高峰。 Kafka 可以開箱即用地執行此操作，但您需要設定一定的間隔。

有一個項目地洞這將為您提供有關 Kafka 的更多資訊。它只是使用消費者組 API 來給出該組的運行狀況。除了 OK 和 Failed 之外，還有一個警告，您可以發現您的消費者無法應對生產速度 - 他們沒有時間校對所寫的內容。該系統非常聰明且易於使用。

這就是 API 回應的樣子。這是群組 bob-live-fifa，分區 returned.update.v1，狀態 OK，滯後 0 - 最後一個最終偏移等等。

監控 Updated_at SLA（卡住） 我已經提過。例如，產品已變更為可以退貨的狀態。我們安裝了 Cron，它表示如果 5 分鐘內該物件還沒有退款（我們很快就透過支付系統回饋資金），那麼肯定出了問題，這絕對是需要支援的情況。因此，我們只需採用 Cron，它會讀取這些內容，如果它們大於 0，則它會發送警報。

總而言之，在以下情況下使用事件很方便:

多個系統需要資訊；
處理的結果並不重要；
事件很少或小。

這篇文章似乎有一個非常具體的主題 - Kafka 上的非同步 API，但與此相關，我想立即推薦很多內容。
首先，接下來高負載++ 我們需要等到 XNUMX 月；XNUMX 月將有聖彼得堡版本，XNUMX 月我們將討論新西伯利亞的高負載。
其次，報告的作者Sergei Zaika是我們新一屆知識管理會議的程序委員會成員知識會議。這次會議為期一天，將於26月XNUMX日舉行，但其議程非常緊張。
並且會在五月 PHP 俄羅斯 и RIT++ （包括 DevOpsConf） - 您也可以在那裡建議您的主題，談論您的經驗並抱怨您的填充錐體。

來源： www.habr.com

在 Kafka 上使用非同步 API 開發退款工具服務的經驗