前言
在很多網際網路應用系統中，請求處理非同步化是提升系統性能一種常用的手段，而基於訊息系統的非同步處理由於具備高可靠性、高吞吐量的特點，因而在併發請求量比較高的網際網路系統中被廣泛應用。與此同時，這種方案也帶來了呼叫鏈路處理上的問題，因為大部分應用請求都會要求同步響應實時處理結果，而由於請求的處理過程已經透過訊息非同步解耦，所以整個呼叫鏈路就變成了非同步鏈路，此時請求鏈路的發起者如何同步拿到響應結果，就需要進行額外的系統設計考慮。
為了更清晰地理解這個問題，小碼哥以最近正在做的共享單車的IOT系統為例，給大家來一張圖描述下，如圖所示：
在上述系統流程中，終端裝置與服務端之間透過MQTT協議相連，而MQTT協議本質上是一種非同步訊息的連線方式，因此業務應用（如圖中的訂單系統）發起開鎖請求後，IOT應用系統會以MQTT協議的方式透過物聯網平臺（此處使用的是AWS IOT服務）向裝置發起開鎖下行訊息，而這一過程在IOT應用系統完成與物聯網平臺的互動後同步呼叫鏈路就結束了，因為物聯網平臺與鎖裝置之間透過MQTT訊息服務非同步解耦了，當然物聯網平臺會透過一系列可靠訊息機制來確保開鎖訊息能夠傳送到指定裝置的監聽MQTT佇列。而至於鎖裝置是否能夠及時接收到開鎖下行MQTT訊息，則取決於鎖裝置當時的行動網路情況。
鎖裝置在收到MQTT開鎖訊息後，會透過嵌入式軟體系統觸發硬體裝置完成開鎖動作，之後就需要透過MQTT上行訊息將開鎖結果反饋到服務端，從而由服務端系統判斷是否建立騎行訂單並計算費用。這一過程需要物聯網平臺監聽指定鎖裝置相應的MQTT上行訊息佇列，由於物聯網平臺需要與上萬個裝置進行連線，因而不可能將每一個鎖裝置所產生的MQTT上行訊息都直接轉發給Iot應用系統，因此在物聯網平臺可以將一類的裝置MQTT訊息轉發至特定的業務訊息佇列中，例如開鎖上行訊息，所有裝置的MQTT開鎖響應上行訊息都可以轉發到表示開鎖響應的Iot業務訊息佇列，如“iot_upstream_lock_response”，這樣Iot業務系統則不需要再關注底層裝置的MQTT訊息，可以用更利於業務理解的方式去處理開鎖響應結果。
現在的問題是透過MQTT協議的開鎖下行訊息、上行訊息已經完全處於兩條不同的非同步網路鏈路，而鏈路的發起者此時卻需要同步等待開鎖結果，但是實際上同步鏈路早已在Iot應用系統向物聯網平臺傳送開鎖訊息後就已經完成，所以為了滿足呼叫方的同步請求/響應需要就需要在Iot應用系統的下發開鎖訊息後進行額外的同步阻塞等待，並監聽開鎖響應的Iot業務訊息佇列“iot_upstream_lock_response”關於此次開鎖請求的上行訊息，之後再結束掉之前的同步阻塞等待邏輯，從而實現向業務呼叫方返回實時開鎖響應結果的同步呼叫效果。那麼在上述流程中如何實施額外的同步阻塞以及如何進行回撥訊息的監聽呢？在接下來的內容中就和大家一起探討具體的實施方案！
解決方案分析
以上問題在使用訊息服務進行非同步解耦的應用場景中是比較普遍的需求，由於非同步呼叫鏈路非常長所以通用的解決思路是在呼叫鏈的起始端進行同步阻塞，而在呼叫鏈的結束端透過回撥的方式來實現，如下圖所示：
在上述圖示中，鏈路起始佇列處在傳送第一次非同步訊息後會開啟一個臨時佇列並同步阻塞監聽該臨時佇列的回撥訊息，而鏈路的結束佇列在完成邏輯處理後需要回調起始佇列監聽的臨時佇列，而由於請求執行緒一直處於阻塞監聽該臨時佇列的狀態，所以一旦收到回撥訊息就可以結束阻塞執行後續流程，從而完成整個鏈路的同步響應。
雖然常見的訊息中介軟體都可以實現以上邏輯，例如小碼哥之前所在的公司就基於RabbitMQ透過臨時佇列的方式實現過訊息鏈路的同步呼叫，但是基於訊息中介軟體的方式多少還是顯得有些繁瑣，對於常見的訊息中介軟體如RocketMQ、RabbitMQ來說非同步訊息才是其強項，如果以大量臨時佇列的建立和銷燬為代價來實現訊息呼叫鏈路的同步，不僅從使用上來說顯得有些麻煩，並且也會對訊息中介軟體的穩定性帶來一些不好的影響。
因此在前面提到的IOT系統中，我們採用了基於Redis的釋出/訂閱功能來實現非同步訊息鏈路的同步化呼叫。而由於Redis的高效能以及Redis的應用場景非常豐富，並且非常適合資料頻繁變動的場景，在系統中既可以作為NoSQL資料庫來使用，同時還支援分散式鎖等功能，因而維護的價效比也相對較高。接下來我們就基於Spring Boot的開發框架來演示如何利用Redis的釋出/訂閱來實現非同步訊息鏈路的同步回撥！
Redis釋出訂閱機制
Redis本身可以透過釋出訂閱機制實現一定的訊息佇列功能，在Redis中透過subscribe/publish等命令可以實現釋出訂閱功能，基於此原先的IOT系統處理示意圖如下：
如上圖所示，在IOT應用端傳送非同步MQTT訊息後會以訊息ID組成的Key作為頻道，並保持請求執行緒對該頻道的同步監聽，直到收到Iot業務訊息佇列的開鎖結果上行訊息後，在訊息佇列的消費端將該上行訊息釋出至同樣以訊息requestId組成的頻道中，從而實現基於Redis釋出訂閱機制的非同步訊息系統同步呼叫效果。
Spring Boot程式碼實現
下面我們基於Spring Boot演示如何透過程式碼進行實現，建立Spring Boot工程後引入Spring Boot Redis整合依賴包，如下：
之後在專案的配置檔案中新增Redis服務連線資訊，如下所示：
此時專案就具備了訪問Redis的能力，接下來我們透過具體的程式碼實現來進行功能演示。訂閱監聽程式碼如下：@RestController
在上述程式碼中，我們模擬了一個開鎖請求，在完成非同步訊息處理後會開啟Redis訂閱監聽，為了實現非同步阻塞還需要我們建立訊息任務物件，程式碼如下：
在訊息任務物件中我們透過JDK1。8新提供的CompletableFuture類實現執行緒阻塞效果，並透過定義訊息監聽物件及超時時間完善處理機制。此外根據Controller層程式碼還需要自定義定義訊息監聽處理物件，程式碼如下：
此時就完成了Redis服務訂閱這部分邏輯的編寫，在後續的邏輯處理中需要完成訊息的釋出才能正常結束此處的阻塞等待，接下來我們寫一段程式碼來模擬訊息釋出，程式碼如下：
此時啟動Spring Boot應用呼叫開鎖模擬介面，邏輯就會暫時處於訂閱等待狀態；之後再模擬呼叫開鎖回撥Redis訊息釋出邏輯，之前的阻塞等待就會因為監聽回撥而完成同步返回。