架構設計|異步請求如何同步處理？

本文創意來自一次業務需求，這次需要接入一個第三方外部服務。由於這個服務只提供異步 API，為了不影響現有系統同步處理的方式，接入該外部服務時，應用對外屏蔽這種差異，內部實現異步請求同步。

全文摘要：

• 異步給現有架構帶來的問題
• Dubbo 異步轉同步解決方法
• 異步轉同步架構設計方案

0x00. 前言

現有一個系統，整體架構如下所示:

這是一個很常見的同步設計方案，上游系統需要等待下游系統接口返回調用結果。

現在需要接入另外一個第三方服務 B，該服務與服務 A 最大區別在於，這是一個異步 API。調用之后，僅僅返回受理成功，處理結果后續通過異步通知返回。

接入之后，整體架構如下所示：

由於網絡隔離策略，通知接收程序與通信服務需要單獨分開部署。若沒此要求，可以將通信服務 B 與通知接收程序合並成一個應用。

另外圖中所有應用采用雙節點部署。

為了不影響 OpenAPI 上游系統同步處理邏輯，通信服務 B 調用第三方服務之后，不能立刻返回，需要等待結果通知，拿到具體返回結果。這就需要通信服務 B 內部將異步轉為同步。

這就是一個典型的異步轉同步問題，整個過程涉及兩個問題。

1. 通信服務 B 業務線程如何進入等待狀態？又如何喚醒正確等待線程？
2. 由於通信服務 B 雙節點部署，通知接收程序如何將結果轉發到正在等待處理的節點？

問題 1 的解決方案參考了 Dubbo 設計思路。

我們在使用 Dubbo 調用遠程服務時，默認情況下，這是一種阻塞式調用方式，即 Consumer 端代碼一直阻塞等待，直到 Provider 端返回為止。

由於 Dubbo 底層基於 Netty 發送網絡請求，這其是一個異步的過程。為了讓業務線程能同步等待，這個過程就需要將異步轉為同步。

0x01. Dubbo 異步轉同步解決辦法

1.1 業務線程同步阻塞

Dubbo 發起遠程調用代碼位於 DubboInvoker#doInvoke：

Dubbo 版本為：2.6.X 版本。2,7.X 重構 DefaultFuture ，但是本質原理還是一樣。

默認情況下，Dubbo 支持同步調用方式，這里將會創建 DefaultFuture 對象。

這里有個非常重要邏輯，每個請求生成一個唯一 ID，然后將 ID 與 DefaultFuture 映射關系，存入 Map 中。

這個請求 ID 在之所以這么重要，是因為消費者並發調用服務發送請求，同時將會有多個業務線程進入阻塞。當收到響應之后，我們需要喚醒正確的等待線程，並將處理結果返回。

通過 ID 這個唯一映射的關系，很自然可以找到其對應 DefaultFuture，喚醒其對應的業務線程。

業務線程調用 DefaultFuture#get方法進入阻塞。這段代碼比較簡單，通過調用 Condition#await 阻塞線層。

1.2 喚醒業務線程

當消費者接收到服務提供者的返回結果，將會調用 DefaultFuture#received 方法。

通過響應對象中的唯一 ID，找到其對應 DefaultFuture 對象，從而將結果設置 DefaultFuture 對象中，然后喚醒的相應的業務線程。

這里實際有個優化點，使用 done#signalAll 代替 done#signal。使用 condition 等待通知機制的時候需要注意這一點。

詳情參考:https://github.com/apache/dubbo/issues/3678

1.3 設計注意點

正常情況下，當消費者接收到響應之后，將會從 FUTURES 這個 Map 移除 DefaultFuture 。

但是在異常情況下，服務提供者若處理緩慢，不能及時返回響應結果，消費者業務線程將會因為超時蘇醒。這種情況下 FUTURES 積壓了無效 DefaultFuture 對象。如果不及時清理，極端情況下，將會發生 OOM 。

DefaultFuture 內部將會開啟一個異步線程，定時輪詢 FUTURES 判斷 DefaultFuture 超時時間，及時清理已經無效（超時）的 DefaultFuture。

0x02. 轉發方案設計

根據 Dubbo 解決思路，問題 1 解決辦法就比較簡單了。具體流程如下：

1. 通信服務 B 內部生成一個唯一請求 ID ，發給第三方服務
2. 若請求成功，內部版使用 Map 存儲對應關系，並使業務線程阻塞等待
3. 通信服務 B 收到異步通知結果，通過 ID 查找對應業務線程，喚醒的相應的線程

這個設計過程需要注意設置合理的超時時間，這個超時時間需要考慮遠程服務調用耗時，可以參考如下公式：

業務線程等待時間=通信服務 B 接口的超時時間 - 調用第三方服務 B 接口消耗時間

這里就不貼出具體的代碼，詳細代碼參考 Dubbo DefaultFuture。

接下來重點看下通知服務如何將結果轉發給正確的通信服務 B 的節點。這里想到兩種方案：

1. SocketServer 方案
2. MQ 方案

2.1 SocketServer

通信服務 B 使用 SocketServer 構建一個服務接收程序，當通知接收程序收到第三方服務 B 通知時，通過 Socket 將結果轉發給通信服務 B。

整個系統架構如下所示：

由於生產服務雙節點部署，通知接收程序就不能寫死轉發地址。這里我們將請求 ID 與通信服務 B socket 服務地址關系存入 Redis 中，然后通知接收程序通過 ID 找到正確的地址。

這個方案說實話有點復雜。

第一 SocketServer 編碼難度較大，編寫一個高效 SocketServer 就比較難，一不小心可能產生各種 Bug。

第二通信服務 B 服務地址配置在配置文件中，由於兩個節點地址不同，這就導致同一應用存在不同配置。這對於后面維護就很不友好。

第三額外引入 Redis 依賴，系統復雜度變高。

2.2 MQ 方案

相對 SocketServer 方案，MQ 方案相對簡單，這里采用 MQ 廣播消費的方式，架構如圖所示：

通知接收程序收到異步通知之后，直接將結果發送到 MQ。

通信服務 B 開啟廣播消費模式，拉取 MQ 消息。

通信服務 B_1 拉取消息，通過請求 ID 映射關系，沒找到內部等待的線程，知道這不是自己的等待消息，於是 B_1 直接丟棄即可。

通信服務 B_2 拉取消息，通過請求 ID 映射關系，順利找到正在等待的線程，然后可以喚醒等待線程，返回最后的結果。

對比 SocketServer 方案，MQ 方案整體流程比較簡單，編程難度低，也沒用存在特殊的配置。

不過這個方案十分依賴 MQ 消息實時性，若 MQ 消息投遞延遲很高，這就會導致通信服務 B 業務線程超時蘇醒，業務異常返回。

這里我們選擇使用 RocketMQ，長輪詢 Pull 方式，可保證消息非常實時，

綜上，這里采用 MQ 的方案。

0x03. 總結

異步轉同步我們需要解決同步阻塞，以及如何喚醒的問題。

阻塞/喚醒可以分別使用 Condition#await/signalAll。不過這個過程我們需要生成一個唯一請求 ID，並且保存這個 ID 與業務線程映射關系。后續等到結果返回我們才能通過唯一 ID 喚醒正確等待線程。

只要了解上面幾點，異步轉同步的問題就就可以迎刃而解。

另外，如果你也有碰到異步轉同步問題，本文的方案希望對你有幫助。如果你有其他設計方案，歡迎留言，一起討論~

參考資料

1. http://dubbo.apache.org/zh-cn/docs/source_code_guide/service-invoking-process.html
2. http://dubbo.apache.org/zh-cn/blog/dubbo-invoke.html

最后說一句 (求關注)

這篇文章其實寫了挺久的，寫的挺難得。之前很早想到寫這篇文章，但是沒想好到底咋寫，艱難產出。

看到這里，點個關注呀，點個贊唄。別下次一定啊，大哥。寫文章很辛苦的，需要來點正反饋。

才疏學淺，難免會有紕漏，如果你發現了錯誤的地方，還請你留言給我指出來，我對其加以修改。

感謝您的閱讀，我堅持原創，十分歡迎並感謝您的關注

歡迎關注我的公眾號：程序通事，獲得日常干貨推送。如果您對我的專題內容感興趣，也可以關注我的博客：studyidea.cn