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

本文創意來自一次業務需求,這次需要接入一個第三方外部服務。由于這個服務只提供異步 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 ,但是本質原理還是一樣。

0082zybply1gc7t3m2louj31650u0u0x

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

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

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

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

來源:Dubbo 官網

業務線程調用 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. //dubbo.apache.org/zh-cn/docs/source_code_guide/service-invoking-process.html
  2. //dubbo.apache.org/zh-cn/blog/dubbo-invoke.html

最后說一句 (求關注)

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

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

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

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

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

posted @ 2020-03-02 07:54  樓下小黑哥  閱讀(...)  評論(...編輯  收藏