最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

Python中HTTP請求體壓縮的實現(xiàn)方法

 更新時間:2026年07月29日 08:27:56   作者:detayun  
日常開發(fā)中使用 Python 調(diào)用 HTTP POST 接口時,如果上傳報文體積較大,未壓縮的數(shù)據(jù)會占用更多帶寬,增加網(wǎng)絡耗時,高并發(fā)場景下還會提升服務器流量壓力,本文講解 HTTP 請求體壓縮原理,提供 requests、httpx、aiohttp 完整實現(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)原理

  1. 將原始報文(JSON/二進制文本)序列化為字節(jié);
  2. 使用 gzip/deflate 算法壓縮字節(jié)數(shù)據(jù);
  3. 請求頭增加 Content-Encoding: gzip;
  4. POST 使用 data / content 參數(shù)傳入壓縮后的二進制,禁止使用 json 參數(shù);
  5. 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 控制臺單行刷新,多行刷新實例

    python 控制臺單行刷新,多行刷新實例

    今天小編就為大家分享一篇python 控制臺單行刷新,多行刷新實例,具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧
    2020-02-02
  • Python應用之利用pyecharts畫中國地圖

    Python應用之利用pyecharts畫中國地圖

    這篇文章主要介紹了Python應用之利用pyecharts畫中國地圖,具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教
    2022-07-07
  • Python實現(xiàn)發(fā)送與接收郵件的方法詳解

    Python實現(xiàn)發(fā)送與接收郵件的方法詳解

    這篇文章主要介紹了Python實現(xiàn)發(fā)送與接收郵件的方法,結合實例形式分析了Python基于smtplib庫使用SMTP協(xié)議進行郵件發(fā)送及基于poplib庫使用POP3服務器接收郵件的相關操作技巧,需要的朋友可以參考下
    2018-03-03
  • Python中的OpenCV圖像閾值處理和平滑處理詳解

    Python中的OpenCV圖像閾值處理和平滑處理詳解

    這篇文章主要介紹了Python中的OpenCV圖像閾值處理和平滑處理詳解,OpenCV是一個跨平臺的計算機視覺庫,可用于開發(fā)實時的圖像處理、計算機視覺以及模式識別程序,需要的朋友可以參考下
    2023-08-08
  • python利用matplotlib庫繪制餅圖的方法示例

    python利用matplotlib庫繪制餅圖的方法示例

    Python強大的原因之一就在于其開源,有很多優(yōu)秀的程序員為其提供了豐富的類庫。Matplotlib就是其中之一,下面這篇文章主要介紹了python如何利用matplotlib庫繪制餅圖的方法示例,有需要的朋友們可以參考借鑒,下面來一起看看吧。
    2016-12-12
  • django中只使用ModleForm的表單驗證

    django中只使用ModleForm的表單驗證

    這篇文章主要介紹了django中只使用ModleForm的表單驗證的相關資料,需要的朋友可以參考下
    2022-12-12
  • Python簡單獲取二維數(shù)組行列數(shù)的方法示例

    Python簡單獲取二維數(shù)組行列數(shù)的方法示例

    這篇文章主要介紹了Python簡單獲取二維數(shù)組行列數(shù)的方法,結合實例形式分析了Python基于numpy模塊的二維數(shù)組相關運算技巧,需要的朋友可以參考下
    2018-12-12
  • django+tornado實現(xiàn)實時查看遠程日志的方法

    django+tornado實現(xiàn)實時查看遠程日志的方法

    今天小編就為大家分享一篇django+tornado實現(xiàn)實時查看遠程日志的方法,具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧
    2019-08-08
  • Python文件讀寫及常用文件的打開方式

    Python文件讀寫及常用文件的打開方式

    這篇文章主要介紹了Python文件讀寫及常用文件的打開方式,文章圍繞主題展開詳細的內(nèi)容介紹,具有一定的參考價值,需要的小伙伴可以參考一下
    2022-09-09
  • Python基礎之函數(shù)嵌套知識總結

    Python基礎之函數(shù)嵌套知識總結

    今天帶大家回顧python基礎知識,文中對Python函數(shù)嵌套作了非常詳細的知識總結,對正在學習python基礎的小伙伴們很有幫助,需要的朋友可以參考下
    2021-05-05

最新評論

铜梁县| 伽师县| 清镇市| 高台县| 新乐市| 西青区| 东乡县| 吉安县| 喀什市| 南澳县| 安新县| 章丘市| 会同县| 延安市| 含山县| 隆回县| 城市| 垫江县| 武定县| 新竹市| 临颍县| 陕西省| 宿州市| 饶河县| 河北区| 板桥市| 灵武市| 安泽县| 孟村| 城口县| 泗水县| 开封县| 商南县| 六安市| 阿克陶县| 九龙城区| 密山市| 南充市| 宜春市| 河北区| 威远县|