Python如何连接WebSocket？-杰瑞科技汇

Python 连接 WebSocket 完全指南：从零到实战，轻松实现实时通信

Meta 描述：

想用 Python 实现实时数据推送、聊天应用或在线游戏？本文是您最需要的 Python WebSocket 连接终极指南，包含原生库、主流框架（FastAPI, Socket.IO）代码示例，详解连接流程、心跳机制与常见问题，助您从入门到精通。

（图片来源网络，侵删）

引言：为什么你需要了解 Python WebSocket？

在当今这个追求实时交互的时代,传统的“请求-响应”模式已无法满足用户对即时性的渴望，无论是股票价格的实时跳动、在线聊天室的秒级消息传递，还是多人协作文档的光标同步，其背后都离不开一项关键技术——WebSocket。

WebSocket 是一种在单个 TCP 连接上进行全双工通信的协议，它允许服务器主动向客户端推送数据，实现了真正意义上的实时通信，而 Python，凭借其简洁的语法和强大的生态，成为了实现 WebSocket 应用的首选语言之一。

本文将带你彻底搞懂如何使用 Python 连接 WebSocket，无论你是想使用原生库进行底层探索，还是借助 FastAPI、Socket.IO 等现代框架快速开发，这里都有你想要的答案。

第一部分：WebSocket 核心概念扫盲

在开始编码前,我们先用一分钟快速理解几个核心概念，这会让你后续的学习事半功倍。

（图片来源网络，侵删）

全双工通信：与 HTTP 的半双工不同，WebSocket 允许数据在客户端和服务器之间同时双向流动，就像一条双向高速公路。
长连接：WebSocket 握手成功后，连接会持续保持，直到客户端或服务器主动关闭，避免了 HTTP 短连接带来的频繁握手开销。
握手：连接建立之初，客户端会发送一个特殊的 HTTP 请求，包含 Upgrade: websocket 头，请求将协议从 HTTP 升级为 WebSocket，服务器同意后，连接便建立。
数据帧：通信的数据被封装在“帧”中发送，这使得 WebSocket 能够处理文本、二进制数据，并支持数据分片。

第二部分：原生 Python 方式：`websockets` 库详解

对于想深入了解 WebSocket 工作原理，或需要轻量级解决方案的开发者来说，使用原生库是最佳选择。websockets 是 Python 社区最流行、最成熟的 WebSocket 客户端和服务端库。

安装 `websockets` 库

pip install websockets

编写一个简单的 WebSocket 服务端

服务端的核心任务是“等待连接”、“接收消息”和“发送消息”。

# server.py
import asyncio
import websockets
# async def define websocket_handler(websocket, path):
#     websocket: 一个 WebSocket 连接对象
#     path: 客户端连接的路径
async def handler(websocket, path):
    print(f"新客户端已连接: {path}")
    try:
        # 无限循环，等待接收来自客户端的消息
        async for message in websocket:
            print(f"收到消息: {message}")
            # 将收到的消息原样返回给客户端 (Echo Server)
            await websocket.send(f"服务器已收到你的消息: {message}")
    except websockets.exceptions.ConnectionClosed as e:
        print(f"客户端 {path} 已断开连接: {e}")
# 启动 WebSocket 服务器
# websockets.serve() 返回一个服务器对象
start_server = websockets.serve(handler, "localhost", 8765)
# 运行服务器直到被停止
asyncio.get_event_loop().run_until_complete(start_server)
print("WebSocket 服务器已启动，监听 ws://localhost:8765")
asyncio.get_event_loop().run_forever()

如何运行： 保存为 server.py，然后在终端执行 python server.py，服务器将在 ws://localhost:8765 上等待连接。

编写一个对应的 WebSocket 客户端

客户端的核心任务是“连接服务器”、“发送消息”和“接收消息”。

（图片来源网络，侵删）

# client.py
import asyncio
import websockets
async def client_logic():
    # 使用 with 语句管理连接，确保连接正确关闭
    async with websockets.connect("ws://localhost:8765") as websocket:
        print("已连接到服务器")
        # 发送一条消息
        await websocket.send("你好，服务器！")
        print("已发送: 你好，服务器！")
        # 等待并接收服务器的回复
        response = await websocket.recv()
        print(f"收到回复: {response}")
# 运行客户端逻辑
asyncio.get_event_loop().run_until_complete(client_logic())

如何运行： 先运行 server.py，再在另一个终端运行 client.py，你将看到服务端和客户端的控制台分别打印出交互信息。

进阶：心跳机制

在长连接中,为了防止因网络中间设备（如 NAT 超时）或一方异常崩溃导致连接“假死”，我们需要心跳机制，客户端定期向服务器发送一个简单的“ping”消息，服务器收到后立即回复一个“pong”，以此确认双方连接正常。

# 在客户端代码中添加心跳逻辑
async def client_with_heartbeat():
    async with websockets.connect("ws://localhost:8765") as websocket:
        print("已连接到服务器")
        # 创建一个任务来发送心跳
        heartbeat_task = asyncio.create_task(send_heartbeat(websocket))
        # 主任务处理业务消息
        try:
            while True:
                response = await websocket.recv()
                print(f"收到回复: {response}")
        except websockets.exceptions.ConnectionClosed:
            print("连接已关闭")
        finally:
            # 取消心跳任务
            heartbeat_task.cancel()
            try:
                await heartbeat_task
            except asyncio.CancelledError:
                pass
async def send_heartbeat(websocket):
    while True:
        try:
            await websocket.send("ping")
            print("已发送心跳: ping")
            await asyncio.sleep(30)  # 每30秒发送一次
        except websockets.exceptions.ConnectionClosed:
            print("心跳因连接关闭而停止")
            break
# 运行客户端
asyncio.get_event_loop().run_until_complete(client_with_heartbeat())

第三部分：现代 Web 框架集成

在实际的 Web 应用中，我们很少直接使用原生库，而是将其集成到 Web 框架中。

FastAPI + `websockets`：构建高性能 API

FastAPI 对 WebSocket 有原生支持，使用起来非常方便。

服务端 (FastAPI):

# fastapi_server.py
from fastapi import FastAPI, WebSocket
from fastapi.responses import HTMLResponse
app = FastAPI()
html = """
<!DOCTYPE html>
<html>
    <head>
        <title>FastAPI WebSocket</title>
    </head>
    <body>
        <h1>WebSocket 聊天室</h1>
        <form action="" onsubmit="sendMessage(event)">
            <input type="text" id="messageText" autocomplete="off"/>
            <button>发送</button>
        </form>
        <ul id='messages'>
        </ul>
        <script>
            var ws = new WebSocket("ws://localhost:8000/ws");
            ws.onmessage = function(event) {
                var messages = document.getElementById('messages')
                var message = document.createElement('li')
                message.textContent = event.data
                messages.appendChild(message)
            };
            function sendMessage(event) {
                var input = document.getElementById("messageText")
                ws.send(input.value)
                input.value = ''
                event.preventDefault()
            }
        </script>
    </body>
</html>
"""
@app.get("/")
async def get():
    return HTMLResponse(html)
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"消息已回显: {data}")
if __name__ == "__main__":
    import uvicorn
    uvicorn.run("fastapi_server:app", host="0.0.0.0", port=8000)

运行后,访问 http://localhost:8000 即可看到一个简单的聊天室页面。

Socket.IO：功能更全面的实时通信库

Socket.IO 是一个库，它可以在 WebSocket 和其他传输方式（如 HTTP 长轮询）之间自动降级，并且内置了房间、命名空间、自动重连等高级功能，非常适合构建复杂的实时应用。

安装： pip install "python-socketio[client]" "python-socketio[asyncio_server]"

服务端 (Socket.IO):

# socketio_server.py
import socketio
# 创建一个异步的 Socket.IO 服务器
sio = socketio.AsyncServer(async_mode='asgi', cors_allowed_origins='*')
app = socketio.ASGIApp(sio)
@sio.event
async def connect(sid, environ):
    print(f"客户端 {sid} 已连接")
    await sio.emit('message', {'data': '欢迎连接！'}, to=sid)
@sio.event
async def disconnect(sid):
    print(f"客户端 {sid} 已断开连接")
@sio.event
async def my_message(sid, data):
    print(f"来自 {sid} 的消息: {data}")
    # 广播消息给所有客户端
    await sio.emit('message', {'data': data}, room=sid)
if __name__ == '__main__':
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

客户端 (Socket.IO - JavaScript): 在 HTML 中引入客户端库，

<script src="https://cdn.socket.io/4.5.4/socket.io.min.js"></script>
<script>
  const socket = io("ws://localhost:8000");
  socket.on("connect", () => {
    console.log("连接成功！");
    socket.emit("my_message", { data: "你好，Python 服务器！" });
  });
  socket.on("message", (data) => {
    console.log("收到消息:", data);
    // 在页面上显示消息
    const messages = document.getElementById('messages');
    const message = document.createElement('li');
    message.textContent = data.data;
    messages.appendChild(message);
  });
</script>

第四部分：常见问题与最佳实践

连接失败？
- 检查地址和端口：确保 ws:// 或 wss:// (WebSocket Secure) 地址和端口正确无误。
- 防火墙：检查本地或服务器的防火墙是否阻止了端口。
- CORS 跨域问题：在 Web 应用中，如果前端和后端域名不同，服务器必须配置 CORS 头，FastAPI 和 Socket.IO 都有便捷的配置方式。

生产环境部署

WS vs WSS：在生产环境中，务必使用 WSS (WebSocket Secure)，它基于 TLS/SSL，可以加密通信内容，防止数据被窃听。
反向代理：通常使用 Nginx 或 Caddy 作为反向代理来处理 WSS，Nginx 需要特殊配置来将 WebSocket 连接从 HTTP/HTTPS 中“升级”出来。

Nginx 配置示例：

server {
    listen 443 ssl;
    server_name yourdomain.com;
    ssl_certificate /path/to/your/cert.pem;
    ssl_certificate_key /path/to/your/key.pem;
    location /ws/ {
        proxy_pass http://localhost:8000; # 你的 Python 应用
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection "upgrade";
        proxy_set_header Host $host;
    }
    # 其他静态文件或 API 代理...
}

如何管理大量连接？
- 使用异步框架（如 FastAPI, Socket.IO 的 ASGI 模式）是基础。
- 考虑使用 Gunicorn/Uvicorn 或 Daphne 来运行多个应用进程。
- 对于超大规模场景,可能需要引入消息队列（如 Redis, RabbitMQ）进行解耦，让 WebSocket 服务器只负责实时推送，由其他业务服务将消息投递到队列中。

从原生的 websockets 库到现代的 FastAPI 和功能强大的 Socket.IO，Python 为我们提供了丰富的工具来轻松实现 WebSocket 实时通信。

如果你想学习原理或需要极致的轻量级，选择 websockets 库。
如果你正在构建一个现代化的 API 服务，FastAPI 是你的不二之选。
如果你的应用需要房间、广播、自动重连等高级特性，Socket.IO 能为你节省大量开发时间。

希望这篇指南能帮助你顺利地使用 Python 搭建起自己的实时应用，就去动手尝试，体验双向通信的魅力吧！

Python如何连接WebSocket？

Python 连接 WebSocket 完全指南：从零到实战，轻松实现实时通信

Meta 描述：

引言：为什么你需要了解 Python WebSocket？

第一部分：WebSocket 核心概念扫盲

第二部分：原生 Python 方式：`websockets` 库详解

安装 `websockets` 库

编写一个简单的 WebSocket 服务端

编写一个对应的 WebSocket 客户端

进阶：心跳机制

第三部分：现代 Web 框架集成

FastAPI + `websockets`：构建高性能 API

Socket.IO：功能更全面的实时通信库

第四部分：常见问题与最佳实践

99ANYc3cd6

Flash CS6视频教程怎么学？新手入门看这里？

笔记本BIOS设置怎么操作？图解教程来了！

Java如何准确获取客户端MAC地址？

linux java 创建文件夹

Flash CS4视频教程怎么学？入门到精通有路径吗？

python phantomjs配置

Java数组为何直接调用toString()输出的是地址？

Java中如何获取cookie？

Python Eclipse插件哪里下载？

Python如何通过import调用shell命令？

Python中initialvalue是什么？如何使用？

Python提示syntaxerror怎么办？

如何安装Python版的protobuf？

Python如何高效判断文件中某一行是否存在？

Mac上还能下载Java SE 6吗？

Pixlr Express教程怎么用？

Python如何连接WebSocket？

Python 连接 WebSocket 完全指南：从零到实战，轻松实现实时通信

Meta 描述：

引言：为什么你需要了解 Python WebSocket？

第一部分：WebSocket 核心概念扫盲

第二部分：原生 Python 方式：websockets 库详解

安装 websockets 库

编写一个简单的 WebSocket 服务端

编写一个对应的 WebSocket 客户端

进阶：心跳机制

第三部分：现代 Web 框架集成

FastAPI + websockets：构建高性能 API

Socket.IO：功能更全面的实时通信库

第四部分：常见问题与最佳实践

相关推荐

python phantomjs配置

第二部分：原生 Python 方式：`websockets` 库详解

安装 `websockets` 库

FastAPI + `websockets`：构建高性能 API