Python中HTTP請求體壓縮的實現(xiàn)方法
前言
日常開發(fā)中使用 Python 調(diào)用 HTTP POST 接口時,如果上傳報文體積較大(大量 JSON、批量數(shù)據(jù)、文本內(nèi)容),未壓縮的數(shù)據(jù)會占用更多帶寬,增加網(wǎng)絡耗時,高并發(fā)場景下還會提升服務器流量壓力。
很多開發(fā)者熟悉服務端返回壓縮響應(Accept-Encoding),卻很少了解:客戶端同樣可以將請求 Body 壓縮后上傳。
本文講解 HTTP 請求體壓縮原理,提供 requests、httpx、aiohttp 完整實現(xiàn)代碼,并梳理常見踩坑點。
關鍵概念區(qū)分
Accept-Encoding: gzip, deflate:告訴服務端,客戶端支持接收壓縮響應(下行壓縮,絕大多數(shù)庫自動支持)Content-Encoding: gzip:告訴服務端,當前 POST 請求 Body 已經(jīng)被壓縮(上行壓縮,需要手動實現(xiàn))
重要前提:服務端必須實現(xiàn)對應解壓邏輯。若后端沒有處理 Content-Encoding 請求頭,直接接收壓縮二進制流會出現(xiàn)解析報錯。
一、實現(xiàn)原理
- 將原始報文(JSON/二進制文本)序列化為字節(jié);
- 使用 gzip/deflate 算法壓縮字節(jié)數(shù)據(jù);
- 請求頭增加
Content-Encoding: gzip; - POST 使用
data/content參數(shù)傳入壓縮后的二進制,禁止使用 json 參數(shù); - HTTP 自動填充
Content-Length,無需手動計算。
推薦算法:gzip,跨語言、各后端框架兼容性最優(yōu)。
二、requests 同步實現(xiàn)(最常用)
import requests
import gzip
import json
from io import BytesIO
def gzip_compress(raw_data: bytes) -> bytes:
"""gzip壓縮字節(jié)數(shù)據(jù)"""
buffer = BytesIO()
with gzip.GzipFile(fileobj=buffer, mode="wb") as fp:
fp.write(raw_data)
return buffer.getvalue()
if __name__ == "__main__":
url = "http://127.0.0.1:8000/api/upload"
payload = {
"list": list(range(2000)),
"content": "大量測試文本" * 500
}
# 序列化為JSON字節(jié)
raw_bytes = json.dumps(payload, ensure_ascii=False).encode("utf-8")
# 閾值判斷:過小的數(shù)據(jù)不壓縮,避免壓縮頭部導致體積變大
compress_threshold = 2048
headers = {
"Content-Type": "application/json; charset=utf-8"
}
if len(raw_bytes) > compress_threshold:
body = gzip_compress(raw_bytes)
headers["Content-Encoding"] = "gzip"
else:
body = raw_bytes
resp = requests.post(url, headers=headers, data=body)
print("狀態(tài)碼:", resp.status_code)
print("響應內(nèi)容:", resp.text)
避坑:不要使用 requests.post(..., json=payload)
json 參數(shù)內(nèi)部會自動編碼,無法傳入壓縮后的二進制數(shù)據(jù),必須使用 data。
三、httpx 同步/異步實現(xiàn)(新項目推薦)
import httpx
import gzip
import json
from io import BytesIO
def gzip_compress(raw_data: bytes) -> bytes:
buffer = BytesIO()
with gzip.GzipFile(fileobj=buffer, mode="wb") as fp:
fp.write(raw_data)
return buffer.getvalue()
def sync_request():
url = "http://127.0.0.1:8000/api/upload"
payload = {"data": list(range(3000))}
raw_bytes = json.dumps(payload).encode("utf-8")
compressed = gzip_compress(raw_bytes)
headers = {
"Content-Type": "application/json; charset=utf-8",
"Content-Encoding": "gzip"
}
with httpx.Client() as client:
r = client.post(url, headers=headers, content=compressed)
print(r.status_code, r.text)
# 異步版本
import asyncio
async def async_request():
url = "http://127.0.0.1:8000/api/upload"
payload = {"data": list(range(3000))}
raw_bytes = json.dumps(payload).encode("utf-8")
compressed = gzip_compress(raw_bytes)
headers = {
"Content-Type": "application/json; charset=utf-8",
"Content-Encoding": "gzip"
}
async with httpx.AsyncClient() as client:
r = await client.post(url, headers=headers, content=compressed)
print(r.status_code, r.text)
if __name__ == "__main__":
sync_request()
# asyncio.run(async_request())
四、aiohttp 異步實現(xiàn)
import aiohttp
import asyncio
import gzip
import json
from io import BytesIO
def gzip_compress(raw_data: bytes) -> bytes:
buffer = BytesIO()
with gzip.GzipFile(fileobj=buffer, mode="wb") as fp:
fp.write(raw_data)
return buffer.getvalue()
async def main():
url = "http://127.0.0.1:8000/api/upload"
payload = {"data": list(range(3000))}
raw_bytes = json.dumps(payload).encode("utf-8")
compressed_body = gzip_compress(raw_bytes)
headers = {
"Content-Type": "application/json; charset=utf-8",
"Content-Encoding": "gzip"
}
async with aiohttp.ClientSession() as session:
async with session.post(url, data=compressed_body, headers=headers) as resp:
print("status:", resp.status)
print(await resp.text())
if __name__ == "__main__":
asyncio.run(main())
五、配套服務端簡易解壓示例(FastAPI)
測試時需要后端支持解壓,附上最簡示例:
from fastapi import FastAPI, Request
import gzip
import json
from io import BytesIO
app = FastAPI()
@app.post("/api/upload")
async def upload(request: Request):
headers = request.headers
body = await request.body()
# 判斷是否gzip壓縮
if headers.get("Content-Encoding") == "gzip":
buf = BytesIO(body)
with gzip.GzipFile(fileobj=buf) as f:
body = f.read()
data = json.loads(body)
return {"msg": "接收成功", "data": data}
六、常見問題與優(yōu)化建議
1. 小數(shù)據(jù)不要強行壓縮
gzip 自帶固定頭部(十幾字節(jié)),報文很小的時候,壓縮后體積可能大于原始數(shù)據(jù)。建議設置閾值(2KB左右),超過閾值再開啟壓縮。
2. Content-Type 不要刪除
Content-Encoding 只是標記編碼方式,不能替代 Content-Type。JSON 請求依舊保留 application/json。
3. 代理、網(wǎng)關兼容性問題
部分老舊反向代理、防火墻不識別 Content-Encoding,可能直接丟棄請求。上線前務必做連通性測試。
4. 區(qū)分上行壓縮與下行壓縮
很多人混淆兩個頭:
Accept-Encoding:客戶端聲明能接收壓縮響應(下載)Content-Encoding:請求體已經(jīng)壓縮(上傳)
requests、httpx 默認自帶 Accept-Encoding,響應自動解壓,無需手動處理。
5. 二進制文件場景
不僅限于 JSON,上傳文本、CSV 等均可使用同樣邏輯;如果是圖片、視頻等本身已壓縮的文件,再次 gzip 收益極低,不建議二次壓縮。
七、適用場景總結
? 適合:批量上報接口、大量JSON文本、日志上報、大數(shù)據(jù)同步
? 不適合:極小報文、圖片/視頻等已壓縮媒體文件
當接口傳輸量大、網(wǎng)絡帶寬有限、跨機房調(diào)用時,開啟請求體壓縮能有效降低傳輸耗時,減少流量開銷。
以上就是Python中HTTP請求體壓縮的實現(xiàn)方法的詳細內(nèi)容,更多關于Python HTTP請求體壓縮的資料請關注腳本之家其它相關文章!
相關文章
Python實現(xiàn)發(fā)送與接收郵件的方法詳解
這篇文章主要介紹了Python實現(xiàn)發(fā)送與接收郵件的方法,結合實例形式分析了Python基于smtplib庫使用SMTP協(xié)議進行郵件發(fā)送及基于poplib庫使用POP3服務器接收郵件的相關操作技巧,需要的朋友可以參考下2018-03-03
Python簡單獲取二維數(shù)組行列數(shù)的方法示例
這篇文章主要介紹了Python簡單獲取二維數(shù)組行列數(shù)的方法,結合實例形式分析了Python基于numpy模塊的二維數(shù)組相關運算技巧,需要的朋友可以參考下2018-12-12
django+tornado實現(xiàn)實時查看遠程日志的方法
今天小編就為大家分享一篇django+tornado實現(xiàn)實時查看遠程日志的方法,具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧2019-08-08

