在這個簡單的教程中，我們將在 Spring Boot 中創建幾個微服務，並透過 Axon 框架組織它們之間的互動。
假設我們有這樣一個任務。
股票市場有交易來源。該來源透過 Rest 介面將交易傳輸給我們。
我們需要接收這些交易，將它們保存在資料庫中並創建方便的記憶體儲存。
該存儲庫必須執行以下功能：
- 返回交易清單；
- 返回完整位置，即表“工具”-“當前證券數量”；
- 返回給定儀器的位置。
我們如何解決這個問題？
根據微服務時尚的準則，我們需要將任務劃分為組件微服務：
- 接收 Rest 交易；
- 將交易保存到資料庫；
- 用於按位置呈現資料的記憶體儲存。
讓我們在本教程的框架內製作第一個和第三個服務，將第二個服務留給第二部分（如果有興趣，請在評論中寫下）。
所以我們有兩個微服務。
第一個接收來自外部的資料。
第二個處理此資料並回應傳入的請求。
當然，我們希望獲得微服務的水平擴展、不間斷更新等優勢。
我們面臨的一個非常艱鉅的任務是什麼？
其實有很多，但現在我們來談談資料如何在這些微服務之間流動。你也可以在他們之間做一個休息，你可以放置某種隊列，你可以想出很多有利弊的東西。
讓我們看看一種可能的方法 - 透過非同步交互 軸突框架.
這樣的解決方案有什麼優點呢？
首先，非同步互動增加了靈活性（是的，這裡有一個缺點，但我們現在只討論優點）。
其次，我們開箱即用 事件溯源 и 連續QRS.
第三，Axon提供了現成的基礎設施，我們只需要專注於業務邏輯的發展。
讓我們開始吧。
我們的項目將在gradle上。它將有三個模組：
- 常見的。具有通用資料結構的模組（我們不喜歡複製貼上）；
- 貿易創造者。具有微服務的模組，用於透過 Rest 接受交易；
- 貿易查詢。具有用於位置顯示的微服務的模組。
讓我們以Spring Boot為基礎，連接Axon啟動器。
Axon 在沒有 Spring 的情況下也能正常工作，但我們將一起使用它們。
這裡我們需要停下來談談Axon。
這是一個客戶端-伺服器系統。有一個伺服器 - 這是一個單獨的應用程序，我們將在 Docker 中運行它。
還有一些客戶端嵌入到微服務中。
這就是我們得到的圖片。首先，Axon 伺服器啟動（在 Docker 中），然後是我們的微服務。
啟動時，微服務會搜尋伺服器並開始與其互動。互動可以分為兩種類型：技術互動和業務互動。
從技術上講，這是「我還活著」訊息的交換（此類訊息可以在偵錯日誌記錄模式中看到）。
商業就是像「新政」這樣的資訊交換。
一個重要的功能：啟動後，微服務可以詢問 Axon 伺服器“發生了什麼”，然後伺服器將累積的事件發送到微服務。這樣微服務就可以相對安全地重啟，不會遺失資料。
透過這種交換方案，我們可以非常輕鬆地啟動許多微服務實例，
並在不同的主機上。
是的，Axon 伺服器的一個實例並不可靠，但目前就是這樣。
我們使用事件溯源和 CQRS 範式。這意味著我們必須有「命令」、「事件」和「選擇」。
我們將有一個命令：「建立交易」、一個事件「交易已建立」和三個選擇：「顯示所有交易」、「顯示頭寸」、「按工具顯示頭寸」。
工作流程如下圖所示：
- tradeCreator 微服務接受 Rest 交易。
- tradeCreator 微服務會建立一個「建立交易」指令並將其傳送到 Axon 伺服器。
- Axon 伺服器接受命令並將命令轉發給有興趣的接收者，在我們的例子中是 tradeCreator 微服務。
- tradeCreator 微服務接收指令，產生「交易建立」事件並將其傳送至 Axon 伺服器。
- Axon 伺服器接收事件並將其轉發給有興趣的訂閱者。
- 目前我們只有一個有興趣的接收者：tradeQueries 微服務。
- tradeQueries 微服務接收事件並更新內部資料。
（重要的是，在事件產生時，tradeQueries 微服務可能無法使用，但一旦啟動，它就會立即收到該事件）。
是的，軸突伺服器是通訊的中心，所有訊息都通過它。
讓我們繼續編碼。
為了不讓程式碼弄亂帖子，下面我將僅提供片段；下面將提供整個範例的連結。
我們先從公共模組common開始。
其中的公共部分是事件（CreatedTradeEvent 類別）。注意名字，其實這是產生這個事件的團隊的名字，不過是過去式。過去，因為首先有一個命令會導致創建一個事件。
其他常見結構包括用於描述部位的類別（Position 類別）、交易（Trade 類別）和交易方（列舉 Side），即購買或出售。
讓我們繼續討論 tradeCreator 模組。
此模組有一個用於接受交易的 Rest 介面（TradeController 類別）。
根據收到的交易，產生「建立交易」命令並將其發送到 axon 伺服器。
@PostMapping("/trade")
public ResponseEntity<String> create(@RequestBody Trade trade) {
var createTradeCommand = CreateTradeCommand.builder()
.tradeId(trade.getTradeId())
...
.build();
var result = commandGateway.sendAndWait(createTradeCommand, 3, TimeUnit.SECONDS);
return ResponseEntity.ok(result.get().toString());
}
TradeAggregate 類別用於處理該命令。
為了讓 Axon 找到它，我們加入了 @Aggregate 註解。
處理命令的方法如下所示（縮寫）：
@CommandHandler
public TradeAggregate(CreateTradeCommand command) {
log.info("command: {}", command);
var event = CreatedTradeEvent.builder()
.tradeId(command.tradeId())
....
.build();
AggregateLifecycle.apply(event);
}
該命令產生一個事件並將其發送到伺服器。
該指令位於 CreateTradeCommand 類別中。
現在讓我們來看看最後一個 tradeQueries 模組。
查詢包中描述了選擇。
該模組還有一個Rest接口
公共類別 TradeController。
例如，讓我們看一下處理請求：「顯示所有交易」。
@GetMapping("/trade/all")
public List<Trade> findAllTrades() {
return queryGateway.query(new FindAllTradesQuery(),
ResponseTypes.multipleInstancesOf(Trade.class)).join();
}
建立獲取請求並將其發送到伺服器。
TradesEventHandler 類別用於處理取得請求。
它有一個標有註釋的方法
@QueryHandler
public List<Position> handleFindCurrentPositionQuery(FindCurrentPositionQuery query)
他負責從記憶體儲存中檢索資料。
問題是如何更新該存儲庫中的信息。
讓我們先了解一下，這只是一組 ConcurrentHashMap，是為特定範例量身定制的。
若要更新它們，請使用以下方法：
@EventHandler
public void on(CreatedTradeEvent event) {
log.info("event:{}", event);
var trade = Trade.builder()
...
.build();
trades.put(event.tradeId(), trade);
position.merge(event.shortName(), event.size(),
(oldValue, value) -> event.side() == Side.BUY ? oldValue + value : oldValue - value);
}它接收“交易創建”事件並更新地圖。
這些都是微服務開發的要點。
您對 Axon 的缺點有何看法？
首先，這是基礎設施的複雜化，出現了一個故障點——Axon伺服器，所有通訊都通過它。
其次，這種分散式系統的缺點表現得非常明顯──數據暫時不一致。在我們的例子中，接收新交易和更新樣本資料之間可能會花費令人無法接受的長時間。
幕後還剩下什麼？
完全沒有提及事件溯源和 CQRS、它是什麼、它的用途。
如果不透露這些概念，有些觀點可能還不清楚。
也許個別程式碼片段也需要解釋。
我們將在以下時間討論這個問題 21九月。
.
來源： www.habr.com
