Python網路請求、JSON轉換、多執行緒、非同步IO

本篇文章將給大家講解Python網路請求、request、非同步請求、JSON、非同步IO（協程）和請求、多執行緒、非同步IO/多程序/多執行緒對比。

網路請求

在 Python 眾多的 HTTP 客戶端中，最有名的莫過於

requests

、

aiohttp

和

httpx

。

在不借助其他第三方庫的情況下，

requests

只能傳送同步請求；

aiohttp

只能傳送非同步請求；

httpx

既能傳送同步請求，又能傳送非同步請求。

那麼怎麼選擇呢

只發同步請求用

requests

，但可配合多執行緒變非同步。

只發非同步請求用

aiohttp

，但可以配合await變同步。

httpx

可以發同步請求也可以非同步，但是請求速度同步略差於

requests

，非同步略差於

aiohttp

Asyncio 的強大。但是，任何一種方案都不是完美的，都存在一定的侷限性，Asyncio 同樣如此。

實際使用中，想要用好 Asyncio，特別是發揮其強大的功能，很多情況下必須得有相應的 Python 庫支援。

比如

requests 庫並不相容 Asyncio，而 aiohttp 庫相容。

requests

這裡先說

requests

安裝依賴

pip install requests

響應

響應的型別：

#獲取介面返回的字串資料r。text#獲取介面返回的json資料，即直接將json格式的資料轉換為json物件r。json（）#獲取介面返回的二進位制資料，假設二進位制資料如果為圖片可以繼續轉換成圖片r。content#獲取原始套接字，使用r。raw請在 requests 請求中加上引數 stream=Truer。raw

獲取請求響應的其他資訊

#獲取狀態碼r。status_code#獲取請求的urlr。url#獲取指定cookies資訊r。cookies［‘token’］#獲取訪問伺服器返回給我們的響應頭部資訊r。headers#獲取指定訪問伺服器返回給我們的響應頭部資訊r。headers［‘Content-Type’］#獲取傳送到伺服器的請求的頭部的資訊r。request。headers

請求

GET請求

get請求：

res = requests。get（url，data=data，cookies=cookie，headers=header，verify=False，files=file）

data可傳可不傳，data是字典格式。

如果url是https的話，加上verify=False。如果url是http的話，可不加。

示例1

import requestsif __name__ == ‘__main__’： r = requests。get（“https：//www。psvmc。cn”） print（r。text）

示例2

import requestsif __name__ == ‘__main__’： data = {‘username’： ‘admin’， ‘passwd’： ‘123456’} r = requests。get（“https：//www。psvmc。cn/login。json”， params=data） print（r。status_code） print（r。json（）［“obj”］）

POST請求

url_post = “https：//www。psvmc。cn/login。json”#不包含任何引數的請求r = requests。post（url_post）#不包含任何引數的請求，設定超時10s，timeout不設定則預設60sr = requests。post（url_post，timeout=10）#攜帶引數的請求，dict_param為引數字典，預設data=dict_param，使用data=則表示post的是form請求#即 application/x-www-form-urlencoded 。r = requests。post（url_post， data=dict_param）#攜帶引數的請求，dict_param為引數字典，使用json=則表示post的是json請求r = requests。post（url_post， json=dict_param）#攜帶引數的請求，body傳字串，這裡是JSON字串。r = requests。post（url_post， data=json。dumps（dict_param））#攜帶引數的請求，dict_param為引數字典，設定超時10s，並攜帶headers屬性r = requests。post（ url_post， data=dict_param， timeout=10， headers={ ‘User-Agent’： ‘Mozilla/5。0 （iPhone； CPU iPhone OS 13_2_3 like Mac OS X）’ }）#post請求上傳檔案url = ‘http：//apihost/upload/post’files = {‘file’： open（‘report。xls’， ‘rb’）}r = requests。post（url， files=files）

其他型別請求

r = requests。put（url， data =dict_param）r = requests。delete（url）r = requests。head（url）r = requests。options（url）

代理

跨域的時候可以考慮代理訪問，不管是post請求還是get請求，只需要新增proxies即可。

客戶端開發時不用考慮跨域問題，沒有必要設定代理訪問。

proxies = { “http”： “http：//10。10。1。10：3128”， “https”： “http：//10。10。1。10：1080”，}requests。get（url_get， proxies=proxies）

檢視代理是否有效

和telnet作用一樣

import telnetlibif __name__ == ‘__main__’： try： telnetlib。Telnet（‘110。242。68。4’， port=‘80’， timeout=3） except： print（‘ip無效！’） else： print（‘ip有效！’）

非同步請求

aiohttp 的程式碼與 httpx 非同步模式的程式碼重合度90%，只不過把

AsyncClient

換成了

ClientSession

，

另外，在使用 httpx 時，當你

await client。post

時就已經發送了請求。但是當使用

aiohttp

時，只有在

awiat resp。json（）

時才會真正傳送請求。

aiohttp

import aiohttpimport asyncioasync def main（）： async with aiohttp。ClientSession（） as client： resp = await client。post（‘https：//www。psvmc。cn/login。json’， json={‘ts’： ‘2020-01-20 13：14：15’}） result = await resp。json（） print（result）asyncio。run（main（））

httpx

import httpximport asyncioasync def main（）： async with httpx。AsyncClient（） as client： resp = await client。post（‘https：//www。psvmc。cn/login。json’， json={‘ts’： ‘2020-01-20 13：14：15’}） result = resp。json（） print（result）asyncio。run（main（））

JSON

字串轉物件

import json# 一些 JSON：x = ‘{ “name”：“Bill”， “age”：63， “city”：“Seatle”}’# 解析 x：y = json。loads（x）# 結果是 Python 字典：print（y［“age”］）

物件轉字串

import json# Python 物件（字典）：x = { “name”： “Bill”， “age”： 63， “city”： “Seatle”}# 轉換為 JSON：y = json。dumps（x）# 結果是 JSON 字串：print（y）

當 Python 轉換為 JSON 時，Python 物件會被轉換為 JSON（JavaScript）等效項：

Python

JSON

dict

Object

list

Array

tuple

Array

str

String

int

Number

float

Number

True

true

False

false

None

null

非同步IO（協程）和請求

pip install aiohttp

簡單示例

import asyncioasync def test（）： await asyncio。sleep（3） return “123”async def main（）： result = await test（） print（result）if __name__ == ‘__main__’： asyncio。run（main（））

非同步請求

import asyncioimport aiohttpimport timeasync def download_one（url）： async with aiohttp。ClientSession（） as session： async with session。get（url） as resp： print（‘Read {} from {}’。format（resp。content_length， url））async def download_all（sites）： tasks = ［asyncio。ensure_future（download_one（site）） for site in sites］ await asyncio。gather（*tasks）def main（）： sites = ［ ‘https：//www。psvmc。cn/index。html’， ‘https：//www。psvmc。cn/login。json’， ‘https：//www。psvmc。cn/userlist。json’ ］ start_time = time。perf_counter（） loop = asyncio。get_event_loop（） try： loop。run_until_complete（download_all（sites）） finally： if loop。is_running（）： loop。close（） end_time = time。perf_counter（） print（‘Download {} sites in {} seconds’。format（ len（sites）， end_time - start_time））if __name__ == ‘__main__’： main（）

多執行緒

from concurrent。futures import ThreadPoolExecutorimport threading# 定義一個準備作為執行緒任務的函式def action（num）： print（threading。current_thread（）。name） return num+100# 建立一個包含4條執行緒的執行緒池with ThreadPoolExecutor（max_workers=3） as pool： future1 = pool。submit（action， 1000） def get_result（future）： print（f“單個任務返回：{future。result（）}”） # 為future1新增執行緒完成的回撥函式 future1。add_done_callback（get_result） print（‘————————————————’） # 使用執行緒執行map計算 results = pool。map（action，（50， 100， 150）） for r in results： print（f“多個任務返回：{r}”）

非同步 IO/多程序/多執行緒對比

非同步 IO（asyncio）、多程序（multiprocessing）、多執行緒（multithreading）

IO 密集型應用CPU等待IO時間遠大於CPU 自身執行時間，太浪費；

常見的 IO 密集型業務包括：瀏覽器互動、磁碟請求、網路爬蟲、資料庫請求等

Python 世界對於 IO 密集型場景的併發提升有 3 種方法：多程序、多執行緒、非同步 IO（asyncio）；

理論上講asyncio是效能最高的，原因如下：

程序、執行緒會有CPU上下文切換

程序、執行緒需要核心態和使用者態的互動，效能開銷大；而協程對核心透明的，只在使用者態執行

程序、執行緒並不可以無限建立，最佳實踐一般是 CPU*2；而協程併發能力強，併發上限理論上取決於作業系統IO多路複用（Linux下是 epoll）可註冊的檔案描述符的極限

那asyncio的實際表現是否如理論上那麼強，到底強多少呢？我構建瞭如下測試場景：

訪問500臺 DB，並sleep 100ms模擬業務查詢

方法 1；順序序列一臺臺執行

方法 2：多程序

方法 3：多執行緒

方法 4：asyncio

方法 5：asyncio+uvloop

最後的

asyncio+uvloop

和官方asyncio 最大不同是用 Cython+libuv 重新實現了asyncio 的事件迴圈（event loop）部分，

官方測試效能是 node。js的 2 倍，持平 golang。

以下測試程式碼需要 Pyhton3。7+：

順序序列一臺臺執行

import recordsuser = “root”pwd = “123456”port = 3306hosts = ［］ # 500臺 db列表def query（host）： conn = records。Database（ f‘mysql+pymysql：//{user}：{pwd}@{host}：{port}/mysql？charset=utf8mb4’） rows = conn。query（‘select sleep（0。1）；’） print（rows［0］）def main（）： for h in hosts： query（h）# main entranceif __name__ == ‘__main__’： main（）

多程序

from concurrent import futuresimport recordsuser = “root”pwd = “123456”port = 3306hosts = ［］ # 500臺 db列表def query（host）： conn = records。Database（ f‘mysql+pymysql：//{user}：{pwd}@{host}：{port}/mysql？charset=utf8mb4’） rows = conn。query（‘select sleep（0。1）；’） print（rows［0］）def main（）： with futures。ProcessPoolExecutor（） as executor： for future in executor。map（query，hosts）： pass# main entranceif __name__ == ‘__main__’： main（）

多執行緒

from concurrent import futuresimport recordsuser = “root”pwd = “123456”port = 3306hosts = ［］ # 500臺 db列表def query（host）： conn = records。Database（ f‘mysql+pymysql：//{user}：{pwd}@{host}：{port}/mysql？charset=utf8mb4’） rows = conn。query（‘select sleep（0。1）；’） print（rows［0］）def main（）： with 。ThreadPoolExecutor（） as executor： for future in executor。map（query，hosts）： pass# main entranceif __name__ == ‘__main__’： main（）

asyncio

import asynciofrom databases import Database user = “root”pwd = “123456”port = 3306hosts = ［］ # 500臺 db列表async def query（host）： DATABASE_URL = f‘mysql+pymysql：//{user}：{pwd}@{host}：{port}/mysql？charset=utf8mb4’ async with Database（DATABASE_URL） as database： query = ‘select sleep（0。1）；’ rows = await database。fetch_all（query=query） print（rows［0］）async def main（）： tasks = ［asyncio。create_task（query（host）） for host in hosts］ await asyncio。gather（*tasks）# main entranceif __name__ == ‘__main__’： asyncio。run（main（））

asyncio+uvloop

import asyncioimport uvloopfrom databases import Databaseuser = “root”pwd = “123456”port = 3306hosts = ［］ # 500臺 db列表async def query（host）： DATABASE_URL = f‘mysql+pymysql：//{user}：{pwd}@{host}：{port}/mysql？charset=utf8mb4’ async with Database（DATABASE_URL） as database： query = ‘select sleep（0。1）；’ rows = await database。fetch_all（query=query） print（rows［0］）async def main（）： tasks = ［asyncio。create_task（query（host）） for host in hosts］ await asyncio。gather（*tasks）# main entranceif __name__ == ‘__main__’： uvloop。install（） asyncio。run（main（））

執行時間對比

方式

執行時間

序列

1m7。745s

多程序

2。932s

多執行緒

4。813s

asyncio

1。068s

asyncio+uvloop