譯者 | 豌豆花下貓@Python貓

聲明 ：本翻譯基于CC BY-NC-SA 4.0【2】授權協議，內容略有改動，轉載請保留原文出處，請勿用于商業(yè)或非法用途。

異步（async）正風靡一時拨与。異步Python、異步Rust艾猜、go买喧、node攀甚、.NET，任選一個你最愛的語言生態(tài)岗喉，它都在使用著一些異步秋度。異步這東西有多好，這在很大程度上取決于語言的生態(tài)及其運行時間钱床，但總體而言荚斯，它有一些不錯的好處。它使得這種事情變得非常簡單：等待可能需要一些時間才能完成的操作查牌。

它是如此簡單事期，以至于創(chuàng)造了無數新的方法來坑人（blow ones foot off）。我想討論的一種情況是纸颜，直到系統出現超載兽泣，你才意識到自己踩到了腳的那一種，這就是背壓（back pressure）管理的主題胁孙。在協議設計中有一個相關術語是流量控制（flow control）唠倦。

什么是背壓

關于背壓的解釋有很多，我推薦閱讀的一個很好的解釋是：Backpressure explained — the resisted flow of data through software【3】涮较。因此稠鼻，與其詳細介紹什么是背壓，我只想對其做一個非常簡短的定義和解釋：背壓是阻礙數據在系統中流通的阻力狂票。背壓聽起來很負面——誰都會想象浴缸因管道堵塞而溢出——但這是為了節(jié)省你的時間候齿。

（譯注：back pressure，除了背壓闺属，還有人譯為“回壓”慌盯、“反壓”）

在這里，我們要處理的東西在所有情況下或多或少都是相同的：我們有一個系統將不同組件組合成一個管道掂器，而該管道需要接收一定數量的傳入消息亚皂。

你可以想象這就像在機場模擬行李運送一樣。行李到達唉匾，經過分類孕讳，裝入飛機匠楚，最后卸下巍膘。在這過程中，一件行李要跟其它行李一起芋簿，被扔進集裝箱進行運輸峡懈。當一個集裝箱裝滿后，需要將其運走与斤。當沒有剩余的集裝箱時肪康，這就是背壓的自然示例〖远瘢現在箱沦，放行李者不能放了倘待，因為沒有集裝箱。

此時必須做出決定熄捍。一種選擇是等待：這通常被稱為排隊（queueing ）或緩沖（buffering）雾狈。另一種選擇是扔掉一些行李廓潜，直到有一個集裝箱到達為止——這被稱為丟棄（dropping）。這聽起來很糟糕善榛，但是稍后我們將探討為什么有時很重要辩蛋。

但是，這里還有另一件事移盆。想象一下悼院，負責將行李放入集裝箱的人在較長的時間內（例如一周）都沒等到集裝箱。最終咒循，如果他們沒有丟棄行李据途，那么他們周圍將有數量龐大的行李。最終叙甸，他們被迫要整理的行李數量太多昨凡，用光了存儲行李的物理空間。到那時蚁署，他們最好是告訴機場便脊，在解決好集裝箱問題之前，不能再接收新的行李了光戈。這通常被稱為流量控制【4】哪痰，是一個至關重要的網絡概念。

通常這些處理管道在每段時間內只能容納一定數量的消息（如本例中的行李箱）久妆。如果數量超過了它晌杰，或者更糟糕的是管道停滯，則可能發(fā)生可怕的事情】晗遥現實世界中的一個例子是倫敦希思羅機場 5 號航站樓開放肋演，由于其 IT 基礎架構無法正常運行，在 10 天內未能完成運送 42,000 件行李烂琴。他們不得不取消 500 多個航班爹殊，并且有一段時間，航空公司決定只允許隨身攜帶行李奸绷。

背壓很重要

我們從希思羅災難中學到的是梗夸，能夠交流背壓至關重要。在現實生活中以及在計算中号醉，時間總是有限的反症。最終人們會放棄等待某些事情辛块。特別是即使某些事物在內部可以永遠等待，但在外部卻不能铅碍。

舉一個現實的例子：如果你的行李需通過倫敦希思羅機場到達目的地巴黎润绵，但是你只能在那呆 7 天，那么如果行李延遲成 10 天到達胞谈，這就毫無意義了授药。實際上，你希望將行李重新路由（re-routed）回你的家鄉(xiāng)機場呜魄。

實際上悔叽，承認失敗（你超負載了）比假裝可運作并持續(xù)保持緩沖狀態(tài)要好爵嗅，因為到了某個時候娇澎，它只會令情況變得更糟。

那么睹晒，為什么在我們編寫了多年的基于線程的軟件時趟庄，背壓都沒有被提出，現在卻突然成為討論的話題呢伪很？有諸多因素的結合戚啥，其中一些因素很容易使人陷入困境。

糟糕的默認方式

為了理解為什么背壓在異步代碼中很重要锉试，我想為你提供一段看似簡單的 Python asyncio 代碼猫十，它展示了一些我們不慎忘記了背壓的情況：

from asyncio import start_server, run

async def on_client_connected(reader, writer):

while True:

data = await reader.readline()

if not data:

break

writer.write(data)

async def server():

srv = await start_server(on_client_connected, '127.0.0.1', 8888)

async with srv:

await srv.serve_forever()

run(server())

如果你剛接觸 async/await 概念，請想象一下在調用 await 的時候呆盖，函數會掛起拖云，直到表達式解析完畢。在這里应又，Python 的 asyncio 庫提供的 start_server 函數會運行一個隱藏的 accept 循環(huán)宙项。它偵聽套接字，并為每個連接的套接字生成一個獨立的任務運行著 on_client_connected 函數株扛。

現在尤筐，這看起來非常簡單明了。你可以刪除所有的 await 和 async 關鍵字洞就，最終的代碼看起來與使用線程方式編寫的代碼非常相似盆繁。

但是，它隱藏了一個非常關鍵的問題奖磁，這是我們所有問題的根源：在某些函數調用的前面沒有 await改基。在線程代碼中繁疤，任何函數都可以 yield咖为。在異步代碼中秕狰，只有異步函數可以。在本例中躁染，這意味著 writer.write 方法無法阻塞鸣哀。那么它是如何工作的呢？它將嘗試將數據直接寫入到操作系統的無阻塞套接字緩沖區(qū)中吞彤。

但是我衬，如果緩沖區(qū)已滿并且套接字會阻塞，會發(fā)生什么饰恕？在用線程的情況下挠羔，我們可以在此處將其阻塞，這很理想埋嵌，因為這意味著我們正在施加一些背壓破加。然而，因為這里沒有線程雹嗦，所以我們不能這樣做范舀。因此，我們只能在此處進行緩沖或者刪除數據了罪。因為刪除數據是非常糟糕的锭环，所以 Python 選擇了緩沖。

現在泊藕，如果有人向其中發(fā)送了很多數據卻沒有讀取辅辩，會發(fā)生什么？好了在那種情況下娃圆，緩沖區(qū)會增大汽久，增大，再增大踊餐。這個 API 缺陷就是為什么 Python 的文檔中說景醇，不要只是單獨使用 write，還要接著寫 drain（譯注：消耗吝岭、排水）：

writer.write(data)

await writer.drain()

drain 會排出緩沖區(qū)上多余的東西三痰。它不會排空整個緩沖區(qū)，只會做到令事情不致失控的程度窜管。那么為什么 write 不做隱式 drain 呢散劫？好吧，這會是一個大規(guī)模的 API 監(jiān)控幕帆，我不確定該如何做到获搏。

這里非常重要的是大多數套接字都基于 TCP，而 TCP 具有內置的流量控制失乾。writer 只會按照 reader 可接受的速度寫入（給予或占用一些緩沖空間）常熙。這對開發(fā)者完全是隱藏的纬乍，因為甚至 BSD 套接字庫都沒有公開這種隱式的流量控制操作。

那我們在這里解決背壓問題了嗎裸卫？好吧仿贬，讓我們看一看在線程世界中會是怎樣。在線程世界中墓贿，我們的代碼很可能會運行固定數量的線程茧泪，而 accept 循環(huán)會一直等待，直到線程變得可用再接管請求聋袋。

然而队伟，在我們的異步示例中，有無數的連接要處理幽勒。這就意味著我們可能收到大量連接缰泡，即使這意味著系統可能會過載。在這個非常簡單的示例中代嗤，可能不成問題棘钞，但請想象一下，如果我們做的是數據庫訪問干毅，會發(fā)生什么宜猜。

想象一個數據庫連接池，它最多提供 50 個連接硝逢。當大多數連接會在連接池處阻塞時姨拥，接受 10000 個連接又有什么用？

等待與等待著等待

好啦渠鸽，終于回到了我最初想討論的地方叫乌。在大多數異步系統中，特別是我在 Python 中遇到的大多數情況中徽缚，即使你修復了所有套接字層的緩沖行為憨奸，也最終會陷入一個將一堆異步函數鏈接在一起，而不考慮背壓的世界凿试。

如果我們以數據庫連接池為例排宰，假設只有 50 個可用連接。這意味著我們的代碼最多可以有 50 個并發(fā)的數據庫會話那婉。假設我們希望處理 4 倍多的請求板甘，因為我們期望應用程序執(zhí)行的許多操作是獨立于數據庫的。一種解決方法是制作一個帶有 200 個令牌的信號量（semaphore）详炬，并在開始時獲取一個盐类。如果我們用完了令牌，就需等待信號量發(fā)放令牌。

但是等一下≡谔現在我們又變成了排隊枪萄！我們只是在更前面排。如果令系統嚴重超負荷硬毕，那么我們會從一開始就一直在排隊呻引。因此礼仗，現在每個人都將等待他們愿意等待的最大時間吐咳，然后放棄。更糟糕的是：服務器可能仍會花一段時間處理這些請求元践，直到它意識到客戶端已消失韭脊，而且不再對響應感興趣。

因此单旁，與其一直等待下去沪羔，我們更希望立即獲得反饋。想象你在一個郵局象浑，并且正在從機器上取票蔫饰，票上會說什么時候輪到你。這張票很好地表明了你需要等待多長時間愉豺。如果等待時間太長篓吁，你會決定棄票走人，以后再來蚪拦。請注意杖剪，你在郵局里的排隊等待時間，與實際處理你的請求的時間無關（例如驰贷，因為有人需要提取包裹盛嘿，檢查文件并采集簽名）。

因此括袒，這是天真的版本次兆，我們只知道自己在等待：

from asyncio.sync import Semaphore

semaphore = Semaphore(200)

async def handle_request(request):

await semaphore.acquire()

try:

return generate_response(request)

finally:

semaphore.release()

對于 handle_request 異步函數的調用者，我們只能看到我們正在等待并且什么都沒有發(fā)生锹锰。我們看不到是因為過載而在等待类垦，還是因為生成響應需花費很長時間而在等待〕切耄基本上蚤认，我們一直在這里緩沖，直到服務器最終耗盡內存并崩潰糕伐。

這是因為我們沒有關于背壓的溝通渠道砰琢。那么我們將如何解決呢？一種選擇是添加一個中間層。現在不幸的是陪汽，這里的 asyncio 信號量沒有用训唱，因為它只會讓我們等待。但是假設我們可以詢問信號量還剩下多少個令牌挚冤，那么我們可以執(zhí)行類似這樣的操作：

from hypothetical_asyncio.sync import Semaphore, Service

semaphore = Semaphore(200)

class RequestHandlerService(Service):

async def handle(self, request):

await semaphore.acquire()

try:

return generate_response(request)

finally:

semaphore.release()

@property

def is_ready(self):

return semaphore.tokens_available()

現在况增，我們對系統做了一些更改。現在训挡，我們有一個 RequestHandlerService澳骤，其中包含了更多信息。特別是它具有了準備就緒的概念澜薄。該服務可以被詢問是否準備就緒为肮。該操作在本質上是無阻塞的，并且是最佳估量肤京。

現在颊艳，調用者會將這個：

response = await handle_request(request)

變成這個：

request_handler = RequestHandlerService()

if not request_handler.is_ready:

response = Response(status_code=503)

else:

response = await request_handler.handle(request)

有多種方法可以完成，但是思想是一樣的忘分。在我們真正著手做某件事之前棋枕，我們有一種方法來弄清楚成功的可能性，如果我們超負荷了妒峦，我們將向上溝通重斑。

現在，我沒有想到如何給這種服務下定義舟山。其設計來自 Rust 的tower【5】和 Rust 的actix-service【6】绸狐。兩者對服務特征的定義都跟它非常相似。

現在累盗，由于它是如此的 racy寒矿，因此仍有可能堆積信號量。現在若债，你可以冒這種風險符相，或者還是在 handle 被調用時就拋出失敗。

一個比 asyncio 更好地解決此問題的庫是 trio蠢琳，它會在信號量上暴露內部計數器啊终，并提供一個 CapacityLimiter，它是對容量限制做了優(yōu)化的信號量傲须，可以防止一些常見的陷阱蓝牲。

數據流和協議

現在，上面的示例為我們解決了 RPC 樣式的情況泰讽。對于每次調用例衍，如果系統過載了昔期，我們會盡早得知。許多協議都有非常直接的方式來傳達“服務器正在加載”的信息佛玄。例如硼一，在 HTTP 中，你可以發(fā)出 503梦抢，并在 header 中攜帶一個 retry-after 字段般贼，它會告知客戶端何時可以重試。在下次重試時會添加一個重新評估的自然點奥吩，判斷是否要使用相同的請求重試哼蛆，或者更改某些內容。例如圈驼，如果你無法在 15 秒內重試人芽，那么最好向用戶顯示這種無能望几，而不是顯示一個無休止的加載圖標绩脆。

但是，請求/響應（request/response）式的協議并不是唯一的協議橄抹。許多協議都打開了持久連接靴迫，讓你傳輸大量的數據。在傳統上楼誓，這些協議中有很多是基于 TCP 的玉锌，如前所述，它具有內置的流量控制疟羹。但是主守，此流量控制并沒有真正通過套接字庫公開，這就是為什么高級協議通常需要向其添加自己的流量控制的原因榄融。例如参淫，在 HTTP2 中，就存在一個自定義流量控制協議愧杯，因為 HTTP2 在單個 TCP 連接上涎才，多路復用多個獨立的數據流（streams）。

因為 TCP 在后臺對流量控制進行靜默式管理力九，這可能會使開發(fā)人員陷入一條危險的道路耍铜，他們只知從套接字中讀取字節(jié)，并誤以為這是所有該知道的信息跌前。但是棕兼，TCP API 具有誤導性，因為從 API 角度來看抵乓，流量控制對用戶完全是隱藏的伴挚。當你設計自己的基于數據流的協議時蹭沛，你需要絕對確保存在雙向通信通道，即發(fā)送方不僅要發(fā)送章鲤，還要讀取摊灭，以查看是否允許它們繼續(xù)發(fā)。

對于數據流败徊，關注點通常是不同的帚呼。許多數據流只是字節(jié)或數據幀的流，你不能僅在它們之間丟棄數據包皱蹦。更糟糕的是：發(fā)送方通常不容易察覺到它們是否應該放慢速度煤杀。在 HTTP2 中，你需要在用戶級別上不斷交錯地讀寫沪哺。你必然要在那里處理流量控制沈自。當你在寫并且被允許寫入時，服務器將向你發(fā)送 WINDOW_UPDATE 幀辜妓。

這意味著數據流代碼變得更為復雜枯途，因為你首先需要編寫一個可以對傳入流量作控制的框架。例如籍滴，hyper-h2【7】Python 庫具有令人驚訝的復雜的文件上傳服務器示例酪夷，【8】該示例基于 curio 的流量控制，但是還未完成孽惰。

新步槍

async/await 很棒晚岭，但是它所鼓勵編寫的內容在過載時會導致災難。一方面是因為它如此容易就排隊勋功，但同時因為在使函數變異步后坦报，會造成 API 損壞。我只能假設這就是為什么 Python 在數據流 writer 上仍然使用不可等待的 write 函數狂鞋。

不過片择，最大的原因是 async/await 使你可以編寫許多人最初無法用線程編寫的代碼。我認為這是一件好事要销，因為它降低了實際編寫大型系統的障礙构回。其缺點是，這也意味著許多以前對分布式系統缺乏經驗的開發(fā)人員現在即使只編寫一個程序疏咐，也遇到了分布式系統的許多問題纤掸。由于多路復用的性質，HTTP2 是一種非常復雜的協議浑塞，唯一合理的實現方法是基于 async/await 的例子借跪。

遇到這些問題的不僅是 async/await 代碼。例如酌壕，Dask【9】是數據科學程序員使用的 Python 并行庫掏愁，盡管沒有使用 async/await歇由，但由于缺乏背壓，【10】仍有一些 bug 報告提示系統內存不足果港。但是這些問題是相當根本的沦泌。

然而，背壓的缺失是一種具有火箭筒大小的步槍辛掠。如果你太晚意識到自己構建了個怪物谢谦，那么在不對代碼庫進行重大更改的情況下，幾乎不可能修復它萝衩，因為你可能忘了在某些本應使用異步的函數上使用異步回挽。

其它的編程環(huán)境對此也無濟于事。人們在所有編程環(huán)境中都遇到了同樣的問題猩谊，包括最新版本的 go 和 Rust千劈。即使在長期開源的非常受歡迎的項目中，找到有關“處理流程控制”或“處理背壓”的開放問題（open issue）也并非罕見牌捷，因為事實證明墙牌，事后添加這一點確實很困難。例如宜鸯，go 從 2014 年起就存在一個開放問題憔古，關于給所有文件系統IO添加信號量遮怜，【11】因為它可能會使主機超載淋袖。aiohttp 有一個問題可追溯到2016年，【12】關于客戶端由于背壓不足而導致破壞服務器锯梁。還有很多很多的例子即碗。

如果你查看 Python 的 hyper-h2文檔，將會看到大量令人震驚的示例陌凳，其中包含類似“不處理流量控制”剥懒、“它不遵守 HTTP/2 流量控制，這是一個缺陷合敦，但在其它方面是沒問題的“初橘，等等。在流量控制一出現的時候充岛，我就認為它非常復雜保檐。很容易假裝這不是個問題，這就是為什么我們會處于這種混亂狀態(tài)的根本原因崔梗。流量控制還會增加大量開銷夜只，并且在基準測試中效果不佳。

那么蒜魄，對于你們這些異步庫開發(fā)人員扔亥，這里給你們一個新年的解決方案：在文檔和 API 中场躯，賦予背壓和流量控制其應得的重視。

相關鏈接

[1] I'm not feeling the async pressure: https://lucumr.pocoo.org/2020/1/1/async-pressure/

[2] CC BY-NC-SA 4.0: https://creativecommons.org/licenses/by-nc-sa/4.0/

[3] Backpressure explained — the resisted flow of data through software: https://medium.com/@jayphelps/backpressure-explained-the-flow-of-data-through-software-2350b3e77ce7

[4] 流量控制: https://en.wikipedia.org/wiki/Flow_control_(data)

[5] tower: https://github.com/tower-rs/tower

[6] actix-service: https://docs.rs/actix-service/

[7] hyper-h2: https://github.com/python-hyper/hyper-h2

[8] 文件上傳服務器示例: https://python-hyper.org/projects/h2/en/stable/curio-example.html

[9] Dask: https://dask.org/

[10] 背壓: https://github.com/dask/distributed/issues/2602

[11] 關于給所有文件系統IO添加信號量: https://github.com/golang/go/issues/7903

[12] 有一個問題可追溯到2016年旅挤，: https://github.com/aio-libs/aiohttp/issues/1368