最近公司 Python 后端項目進行重構(gòu)，整個后端邏輯基本都變更為采用“異步”協(xié)程的方式實現(xiàn)。看著滿屏幕經(jīng)過 async await（協(xié)程在 Python 中的實現(xiàn)）修飾的代碼，我頓時感到一臉懵逼，不知所措。

雖然之前有了解過“協(xié)程”是什么東西，但并沒有深入探索，于是正好借著這次機會可以好好學(xué)習(xí)一下。

什么是協(xié)程？

簡單來說，協(xié)程是一種基于線程之上，但又比線程更加輕量級的存在。對于系統(tǒng)內(nèi)核來說，協(xié)程具有不可見的特性，所以這種由 程序員自己寫程序來管理 的輕量級線程又常被稱作 “用戶空間線程”。

協(xié)程比多線程好在哪呢？

1. 線程的控制權(quán)在操作系統(tǒng)手中，而 協(xié)程的控制權(quán)完全掌握在用戶自己手中，因此利用協(xié)程可以減少程序運行時的上下文切換，有效提高程序運行效率。

2. 建立線程時，系統(tǒng)默認(rèn)分配給線程的 棧 大小是 1 M，而協(xié)程更輕量，接近 1 K 。因此可以在相同的內(nèi)存中開啟更多的協(xié)程。

3. 由于協(xié)程的本質(zhì)不是多線程而是單線程，所以不需要多線程的鎖機制。因為只有一個線程，也不存在同時寫變量而引起的沖突。在協(xié)程中控制共享資源不需要加鎖，只需要判斷狀態(tài)即可。所以協(xié)程的執(zhí)行效率比多線程高很多，同時也有效避免了多線程中的競爭關(guān)系。

協(xié)程的適用 & 不適用場景

適用場景：協(xié)程適用于被阻塞的，且需要大量并發(fā)的場景。

不適用場景：協(xié)程不適用于存在大量計算的場景（因為協(xié)程的本質(zhì)是單線程來回切換），如果遇到這種情況，還是應(yīng)該使用其他手段去解決。

初探異步 http 框架 httpx

至此我們對 “協(xié)程” 應(yīng)該有了個大概的了解，但故事說到這里，相信有朋友還是滿臉疑問：“協(xié)程” 對于接口測試有什么幫助呢？不要著急，答案就在下面。

相信用過 Python 做接口測試的朋友都對 requests 庫不陌生。requests 中實現(xiàn)的 http 請求是同步請求，但其實基于 http 請求 IO 阻塞的特性，非常適合用協(xié)程來實現(xiàn) “異步” http 請求從而提升測試效率。

相信早就有人注意到了這點，于是在 Github 經(jīng)過了一番探索后，果不其然，最終尋找到了支持協(xié)程 “異步” 調(diào)用 http 的開源庫： httpx

什么是 httpx

httpx 是一個幾乎繼承了所有 requests 的特性并且支持 “異步” http 請求的開源庫。簡單來說，可以認(rèn)為 httpx 是強化版 requests。

下面大家可以跟著我一起見識一下 httpx 的強大

安裝

httpx 的安裝非常簡單，在 Python 3.6 以上的環(huán)境執(zhí)行

pip install httpx

最佳實踐

俗話說得好，效率決定成敗。我分別使用了 httpx 異步 和 同步 的方式對批量 http 請求進行了耗時比較，來一起看看結(jié)果吧～

首先來看看同步 http 請求的耗時表現(xiàn)：

import asyncioimport httpximport threadingimport time

def sync_main（url， sign）： response = httpx.get（url）.status_code print（f‘sync_main： {threading.current_thread（）}： {sign}： {response}’）

sync_start = time.time（）［sync_main（url=‘http://www.baidu.com’， sign=i） for i in range（200）］sync_end = time.time（）print（sync_end - sync_start）

代碼比較簡單，可以看到在 sync_main 中則實現(xiàn)了同步 http 訪問百度 200 次。

運行后輸出如下（截取了部分關(guān)鍵輸出。。。）：

sync_main： 《_MainThread（MainThread， started 4471512512）》： 192： 200sync_main： 《_MainThread（MainThread， started 4471512512）》： 193： 200sync_main： 《_MainThread（MainThread， started 4471512512）》： 194： 200sync_main： 《_MainThread（MainThread， started 4471512512）》： 195： 200sync_main： 《_MainThread（MainThread， started 4471512512）》： 196： 200sync_main： 《_MainThread（MainThread， started 4471512512）》： 197： 200sync_main： 《_MainThread（MainThread， started 4471512512）》： 198： 200sync_main： 《_MainThread（MainThread， started 4471512512）》： 199： 20016.56578803062439

可以看到在上面的輸出中， 主線程沒有進行切換（因為本來就是單線程啊喂?。┱埱蟀凑枕樞驁?zhí)行（因為是同步請求）。

程序運行共耗時 16.6 秒

下面我們試試 “異步” http 請求：

import asyncioimport httpximport threadingimport time

client = httpx.AsyncClient（）

async def async_main（url， sign）： response = await client.get（url） status_code = response.status_code print（f‘a(chǎn)sync_main： {threading.current_thread（）}： {sign}：{status_code}’）

loop = asyncio.get_event_loop（）tasks = ［async_main（url=‘http://www.baidu.com’， sign=i） for i in range（200）］async_start = time.time（）loop.run_until_complete（asyncio.wait（tasks））async_end = time.time（）loop.close（）print（async_end - async_start）

上述代碼在 async_main 中用 async await 關(guān)鍵字實現(xiàn)了“異步” http，通過 asyncio （ 異步 io 庫請求百度首頁 200 次并打印出了耗時。

運行代碼后可以看到如下輸出（截取了部分關(guān)鍵輸出。。。）

async_main： 《_MainThread（MainThread， started 4471512512）》： 56： 200async_main： 《_MainThread（MainThread， started 4471512512）》： 99： 200async_main： 《_MainThread（MainThread， started 4471512512）》： 67： 200async_main： 《_MainThread（MainThread， started 4471512512）》： 93： 200async_main： 《_MainThread（MainThread， started 4471512512）》： 125： 200async_main： 《_MainThread（MainThread， started 4471512512）》： 193： 200async_main： 《_MainThread（MainThread， started 4471512512）》： 100： 2004.518340110778809

可以看到順序雖然是亂的（56，99，67.。。） （這是因為程序在協(xié)程間不停切換） 但是主線程并沒有切換 （協(xié)程本質(zhì)還是單線程 ）。

程序共耗時 4.5 秒

比起同步請求耗時的 16.6 秒 縮短了接近 73 %！

俗話說得好，一步快，步步快。 在耗時方面，“異步” http 確實比同步 http 快了很多。當(dāng)然，“協(xié)程” 不僅僅能在請求效率方面賦能接口測試， 掌握 “協(xié)程”后，相信小伙伴們的技術(shù)水平也能提升一個臺階，從而設(shè)計出更優(yōu)秀的測試框架。

編輯：jq