Python实现长连接的多种方式解析

2025-12-18 08:44:50 作者：破烂pan

本文介绍了Python实现长连接的多种方式,包括HTTP长连接、WebSocket、Server-Sent Events、TCPSocket、异步长连接模型和消息系统模拟长连接,每种方式都有其适用场景和优缺点,并提供了示例代码和生产注意事项,需要的朋友可以参考下

一、什么是“长连接”（Long-lived Connection）

长连接指客户端与服务端建立连接后，在较长时间内保持不关闭，用于多次通信或持续数据流传输。

核心目标：

降低频繁建连/断连的开销（TCP / TLS）
实现实时通信、流式数据、推送能力
支撑高并发、低延迟业务场景

二、Python 实现长连接的主流方式总览

从 通信层级 来看，Python 的长连接可以分为：

层级	技术
HTTP 层	Keep-Alive / SSE
应用协议层	WebSocket
传输层	TCP Socket
异步模型	asyncio / uvloop
消息系统	Redis Pub/Sub 等

下面逐一展开。

三、HTTP 长连接（Keep-Alive）

1. 原理说明

HTTP/1.1 默认开启 Connection: keep-alive
同一个 TCP 连接可复用多次 HTTP 请求。

注意：
HTTP 长连接 ≠ 实时通信
它只是“连接复用”，不是推送模型。

2. 适用场景

内部 API 调用
微服务间 RPC
高频短请求（如模型推理接口）

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()

原理点

Session 内部维护连接池
复用 TCP 连接，减少握手开销

4. 优缺点分析

优点

实现简单
与现有 HTTP 服务完全兼容

缺点

不支持服务端主动推送
连接空闲易被 LB / Nginx 断开

5. 生产注意事项

Nginx keepalive_timeout
服务端最大连接数
不适合实时推送类业务

四、WebSocket（全双工长连接）

1. 原理说明

WebSocket 基于 HTTP 握手升级协议：

HTTP → Upgrade → WebSocket

特点：

全双工
服务端可主动推送
一个连接可持续通信

2. 典型使用场景

实时消息 / IM
模型流式输出（Token Streaming）
实时监控面板
在线协作

3. FastAPI WebSocket 示例

服务端

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()

客户端测试脚本

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. 优缺点分析

优点

真正的实时通信
支持流式数据

缺点

状态连接，消耗资源
LB 场景需粘性会话

5. 生产级注意事项

心跳（ping/pong）
最大连接数控制
Nginx proxy_read_timeout
Kubernetes 下需 Session Affinity

五、Server-Sent Events（SSE）

1. 原理说明

基于 HTTP
单向：服务端 → 客户端
使用 text/event-stream

2. 适用场景

实时通知
日志流
AI 模型流式输出（无双向需求）

3. FastAPI SSE 示例

服务端

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")

客户端测试脚本

import requests

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

4. 优缺点

优点

比 WebSocket 简单
HTTP 原生，穿透性好

缺点

单向通信
浏览器连接数限制

六、TCP Socket 长连接（底层方案）

1. 原理说明

直接基于 TCP
自定义协议
需自行处理粘包、心跳、重连

2. asyncio TCP 示例

服务端

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. 适用场景

游戏服务器
自定义高性能协议
内部 RPC

4. 风险点

实现成本高
协议设计复杂
调试困难

七、异步长连接模型（asyncio / uvloop）

核心价值

单线程支撑万级连接
IO 多路复用（epoll）

import asyncio
import uvloop

asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

结论：
生产级长连接，几乎必用 asyncio

八、消息系统模拟“长连接”

Redis Pub/Sub 示例

import redis

r = redis.Redis()

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

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

特点

解耦
非真实 TCP 长连接
更适合集群广播

九、Python 如何测试长连接（重点）

1. 长时间保持连接

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

2. 心跳机制

WebSocket ping/pong
TCP 自定义 heartbeat
超时主动关闭

3. 并发测试（asyncio）

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

十、生产级最佳实践总结

什么时候用长连接？

✅ 实时
✅ 高频交互
✅ 流式输出

什么时候不用？

❌ 请求极少
❌ LB 无法保持会话
❌ 资源极度受限

十一、方案对比表

技术	通信模型	推送	并发	复杂度	推荐
HTTP Keep-Alive	请求/响应	❌	高	低	⭐⭐⭐
WebSocket	全双工	✅	高	中	⭐⭐⭐⭐⭐
SSE	单向流	✅	中	低	⭐⭐⭐⭐
TCP Socket	自定义	✅	极高	高	⭐⭐⭐
Redis Pub/Sub	消息	间接	高	中	⭐⭐⭐⭐

十二、最终结论

**Python 生产级长连接首选：
asyncio + WebSocket（或 SSE）
TCP 仅在极致性能或特殊协议下使用

以上就是Python实现长连接的多种方式解析的详细内容，更多关于Python实现长连接的资料请关注脚本之家其它相关文章！

Python实现长连接的多种方式解析

一、什么是“长连接”（Long-lived Connection）

二、Python 实现长连接的主流方式总览

三、HTTP 长连接（Keep-Alive）

1. 原理说明

2. 适用场景

3. 示例代码（requests + Session）

原理点

4. 优缺点分析

5. 生产注意事项

四、WebSocket（全双工长连接）

1. 原理说明

2. 典型使用场景

3. FastAPI WebSocket 示例

服务端

客户端测试脚本

4. 优缺点分析

5. 生产级注意事项

五、Server-Sent Events（SSE）

1. 原理说明

2. 适用场景

3. FastAPI SSE 示例

服务端

客户端测试脚本

4. 优缺点

六、TCP Socket 长连接（底层方案）

1. 原理说明

2. asyncio TCP 示例

服务端

客户端

3. 适用场景

4. 风险点

七、异步长连接模型（asyncio / uvloop）

核心价值

八、消息系统模拟“长连接”

Redis Pub/Sub 示例

特点

九、Python 如何测试长连接（重点）

1. 长时间保持连接

2. 心跳机制

3. 并发测试（asyncio）

十、生产级最佳实践总结

什么时候用长连接？

什么时候不用？

十一、方案对比表

十二、最终结论

您可能感兴趣的文章: