一、什么是“長(zhǎng)連接”（Long-lived Connection）

長(zhǎng)連接指客戶端與服務(wù)端建立連接后，在較長(zhǎng)時(shí)間內(nèi)保持不關(guān)閉，用于多次通信或持續(xù)數(shù)據(jù)流傳輸。

核心目標(biāo)：

• 降低頻繁建連/斷連的開銷（TCP / TLS）
• 實(shí)現(xiàn)實(shí)時(shí)通信、流式數(shù)據(jù)、推送能力
• 支撐高并發(fā)、低延遲業(yè)務(wù)場(chǎng)景

二、Python 實(shí)現(xiàn)長(zhǎng)連接的主流方式總覽

從 通信層級(jí) 來看，Python 的長(zhǎng)連接可以分為：

下面逐一展開。

三、HTTP 長(zhǎng)連接（Keep-Alive）

1. 原理說明

HTTP/1.1 默認(rèn)開啟 Connection: keep-alive
同一個(gè) TCP 連接可復(fù)用多次 HTTP 請(qǐng)求。

注意：
HTTP 長(zhǎng)連接 ≠ 實(shí)時(shí)通信
它只是“連接復(fù)用”，不是推送模型。

2. 適用場(chǎng)景

• 內(nèi)部 API 調(diào)用
• 微服務(wù)間 RPC
• 高頻短請(qǐng)求（如模型推理接口）

3. 示例代碼（requests + Session）

import requests

session = requests.Session()

for i in range(10):
    resp = session.get("http://localhost:8000/health")
    print(resp.status_code)

session.close()

原理點(diǎn)

• Session 內(nèi)部維護(hù)連接池
• 復(fù)用 TCP 連接，減少握手開銷

4. 優(yōu)缺點(diǎn)分析

優(yōu)點(diǎn)

• 實(shí)現(xiàn)簡(jiǎn)單
• 與現(xiàn)有 HTTP 服務(wù)完全兼容

缺點(diǎn)

• 不支持服務(wù)端主動(dòng)推送
• 連接空閑易被 LB / Nginx 斷開

5. 生產(chǎn)注意事項(xiàng)

• Nginx keepalive_timeout
• 服務(wù)端最大連接數(shù)
• 不適合實(shí)時(shí)推送類業(yè)務(wù)

四、WebSocket（全雙工長(zhǎng)連接）

1. 原理說明

WebSocket 基于 HTTP 握手升級(jí)協(xié)議：

HTTP → Upgrade → WebSocket

特點(diǎn)：

• 全雙工
• 服務(wù)端可主動(dòng)推送
• 一個(gè)連接可持續(xù)通信

2. 典型使用場(chǎng)景

• 實(shí)時(shí)消息 / IM
• 模型流式輸出（Token Streaming）
• 實(shí)時(shí)監(jiān)控面板
• 在線協(xié)作

3. FastAPI WebSocket 示例

服務(wù)端

from fastapi import FastAPI, WebSocket
import asyncio

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
    await ws.accept()
    try:
        while True:
            data = await ws.receive_text()
            await ws.send_text(f"echo: {data}")
    except Exception:
        await ws.close()

客戶端測(cè)試腳本

import asyncio
import websockets

async def test_ws():
    async with websockets.connect("ws://localhost:8000/ws") as ws:
        await ws.send("hello")
        resp = await ws.recv()
        print(resp)

asyncio.run(test_ws())

4. 優(yōu)缺點(diǎn)分析

優(yōu)點(diǎn)

• 真正的實(shí)時(shí)通信
• 支持流式數(shù)據(jù)

缺點(diǎn)

• 狀態(tài)連接，消耗資源
• LB 場(chǎng)景需粘性會(huì)話

5. 生產(chǎn)級(jí)注意事項(xiàng)

• 心跳（ping/pong）
• 最大連接數(shù)控制
• Nginx proxy_read_timeout
• Kubernetes 下需 Session Affinity

五、Server-Sent Events（SSE）

1. 原理說明

• 基于 HTTP
• 單向：服務(wù)端 → 客戶端
• 使用 text/event-stream

2. 適用場(chǎng)景

• 實(shí)時(shí)通知
• 日志流
• AI 模型流式輸出（無雙向需求）

3. FastAPI SSE 示例

服務(wù)端

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import time

app = FastAPI()

def event_generator():
    for i in range(5):
        yield f"data: message {i}\n\n"
        time.sleep(1)

@app.get("/sse")
def sse():
    return StreamingResponse(event_generator(),
                             media_type="text/event-stream")

客戶端測(cè)試腳本

import requests

resp = requests.get("http://localhost:8000/sse", stream=True)
for line in resp.iter_lines():
    if line:
        print(line.decode())

4. 優(yōu)缺點(diǎn)

優(yōu)點(diǎn)

• 比 WebSocket 簡(jiǎn)單
• HTTP 原生，穿透性好

缺點(diǎn)

• 單向通信
• 瀏覽器連接數(shù)限制

六、TCP Socket 長(zhǎng)連接（底層方案）

1. 原理說明

• 直接基于 TCP
• 自定義協(xié)議
• 需自行處理粘包、心跳、重連

2. asyncio TCP 示例

服務(wù)端

import asyncio

async def handle(reader, writer):
    while True:
        data = await reader.read(1024)
        if not data:
            break
        writer.write(data)
        await writer.drain()

asyncio.run(asyncio.start_server(handle, "0.0.0.0", 9000))

客戶端

import asyncio

async def client():
    reader, writer = await asyncio.open_connection("127.0.0.1", 9000)
    writer.write(b"ping")
    await writer.drain()
    data = await reader.read(1024)
    print(data)
    writer.close()

asyncio.run(client())

3. 適用場(chǎng)景

• 游戲服務(wù)器
• 自定義高性能協(xié)議
• 內(nèi)部 RPC

4. 風(fēng)險(xiǎn)點(diǎn)

• 實(shí)現(xiàn)成本高
• 協(xié)議設(shè)計(jì)復(fù)雜
• 調(diào)試?yán)щy

七、異步長(zhǎng)連接模型（asyncio / uvloop）

核心價(jià)值

• 單線程支撐萬級(jí)連接
• IO 多路復(fù)用（epoll）

import asyncio
import uvloop

asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

結(jié)論：
生產(chǎn)級(jí)長(zhǎng)連接，幾乎必用 asyncio

八、消息系統(tǒng)模擬“長(zhǎng)連接”

Redis Pub/Sub 示例

import redis

r = redis.Redis()

pubsub = r.pubsub()
pubsub.subscribe("events")

for msg in pubsub.listen():
    print(msg)

特點(diǎn)

• 解耦
• 非真實(shí) TCP 長(zhǎng)連接
• 更適合集群廣播

九、Python 如何測(cè)試長(zhǎng)連接（重點(diǎn)）

1. 長(zhǎng)時(shí)間保持連接

while True:
    ws.send("ping")
    time.sleep(30)

2. 心跳機(jī)制

• WebSocket ping/pong
• TCP 自定義 heartbeat
• 超時(shí)主動(dòng)關(guān)閉

3. 并發(fā)測(cè)試（asyncio）

async def run(n=1000):
    tasks = [test_ws() for _ in range(n)]
    await asyncio.gather(*tasks)

十、生產(chǎn)級(jí)最佳實(shí)踐總結(jié)

什么時(shí)候用長(zhǎng)連接？

? 實(shí)時(shí)
? 高頻交互
? 流式輸出

什么時(shí)候不用？

? 請(qǐng)求極少
? LB 無法保持會(huì)話
? 資源極度受限

十一、方案對(duì)比表

十二、最終結(jié)論

**Python 生產(chǎn)級(jí)長(zhǎng)連接首選：
asyncio + WebSocket（或 SSE）
TCP 僅在極致性能或特殊協(xié)議下使用

以上就是Python實(shí)現(xiàn)長(zhǎng)連接的多種方式解析的詳細(xì)內(nèi)容，更多關(guān)于Python實(shí)現(xiàn)長(zhǎng)連接的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！