FastAPI WebSocket并发处理入门教程：让Python新手轻松上手实时应用

欢迎来到Web开发的世界！

如果你是Python新手，刚学会基础语法，但还没接触过Web开发，别担心！本教程将用最简单的方式，带你走进FastAPI的世界，并教你如何处理WebSocket并发，就像学骑自行车一样，一步步来，保证你快速上手！

为什么选FastAPI？

想象一下，FastAPI就像是为你准备的“速成工具包”，它基于Python，代码简洁，文档丰富，特别适合初学者。它不仅能快速搭建API，还内置了WebSocket支持，让你轻松处理实时数据，比如在线聊天、股票报价等。

第一部分：先热个身——快速搭建你的第一个FastAPI应用

在深入WebSocket之前，我们先从基础开始，确保你有“成功时刻”的体验。

步骤1：安装FastAPI

在命令行中运行以下命令，就像安装一个Python包一样简单：

pip install fastapi uvicorn

Uvicorn是一个ASGI服务器，就像Web应用的“发动机”，让FastAPI跑起来。

步骤2：创建一个简单的API

创建一个新文件，比如 main.py，并写入以下代码：

from fastapi import FastAPI

app = FastAPI()  # 创建一个FastAPI应用，就像给你的应用起个名字

@app.get("/")  # 定义一个GET路由，就像设置一个欢迎页面
def read_root():
    return {"Hello": "World"}  # 返回一个JSON响应，简单吧！

运行应用：

uvicorn main:app --reload

现在，打开浏览器访问 http://127.0.0.1:8000，你会看到 {"Hello": "World"}！🎉 恭喜，你的第一个Web应用运行成功了！

成功时刻：看到页面显示JSON数据，你已经迈出了第一步！

第二部分：认识WebSocket——就像一场实时的电话聊天

WebSocket不同于传统的HTTP请求（像发送邮件），它允许服务器和客户端双向实时通信，就像打电话一样，双方可以随时说话。这在需要即时更新的应用中非常有用。

生活化类比

• HTTP请求：你给朋友发一封邮件，等几个小时才回复。
• WebSocket连接：你和朋友打视频电话，可以实时对话，看到对方的表情。

步骤3：添加一个简单的WebSocket端点

更新 main.py，加入WebSocket支持：

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

@app.websocket("/ws")  # 定义一个WebSocket端点，路径是/ws
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()  # 接受连接，就像接起电话
    while True:
        data = await websocket.receive_text()  # 接收消息
        await websocket.send_text(f"Message received: {data}")  # 发送回复
        # 简单示例：客户端发什么，服务器就回什么

解释：这里，async 和 await 是Python的异步关键字，让处理更高效，但作为初学者，先不用深究——想象成让代码“等待”一下，避免卡住。

运行应用，然后你可以用WebSocket客户端测试，比如安装 websockets 库并写一个简单脚本，但为了快速体验，我们可以先用在线工具或浏览器扩展。

成功时刻：如果测试连接，发送消息后收到回复，你就实现了实时通信！

第三部分：处理WebSocket并发——就像举办一场派对，同时和多人聊天

并发处理指的是服务器能同时处理多个WebSocket连接，不会因为一个连接忙而卡住其他。在FastAPI中，由于它是异步的，并发处理天生就很简单。

生活化类比

• 单线程处理：你一次只能和一个人聊天，其他人得等着。
• 并发处理（异步）：你像派对主人，同时和几组人聊天，快速切换话题，每个人都不觉得被忽略。

步骤4：演示并发处理

FastAPI使用异步框架，所以多个WebSocket连接会自动并发处理。我们修改示例来模拟这个：

更新 main.py：

from fastapi import FastAPI, WebSocket
import asyncio

app = FastAPI()

# 存储所有连接的WebSocket，用于广播消息
connections = []

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    connections.append(websocket)  # 添加新连接到列表
    try:
        while True:
            data = await websocket.receive_text()
            # 广播消息给所有连接的用户
            for connection in connections:
                await connection.send_text(f"User said: {data}")
    except Exception:
        connections.remove(websocket)  # 如果连接断开，移除它

# 添加一个简单的页面来测试多个连接
@app.get("/chat")
async def chat_page():
    # 返回一个简单HTML页面，用户可以在浏览器打开多个标签页模拟并发
    return {"message": "Open multiple browser tabs to /ws path to test concurrency!"}

解释：这里，connections 列表存储所有活跃的WebSocket连接，当有消息时，广播给所有人。由于FastAPI是异步的，它可以同时处理多个 websocket_endpoint 实例，实现并发。

运行应用，访问 http://127.0.0.1:8000/chat 查看提示。然后，在浏览器中打开多个标签页，连接到WebSocket（例如使用开发者工具或扩展），发送消息——你会看到所有标签页都收到广播！

成功时刻：打开多个连接并看到消息同步更新，这证明了并发处理在工作！

总结与下一步

恭喜你！通过这个教程，你已经学会了：

1. 搭建基础的FastAPI应用。
2. 实现简单的WebSocket实时通信。
3. 理解并发处理的概念，并看到它在FastAPI中的简单实现。

FastAPI让这些变得非常容易，尤其是对初学者。记住，Web开发不是魔法，而是一步步的实践。

动手挑战

尝试修改代码，比如：

• 添加用户名字，让广播消息显示是谁发的。
• 限制连接数或实现私聊功能。

延伸学习

• 深入学习FastAPI文档：了解更高级的路由、依赖注入等。
• 探索异步编程：学习 asyncio 来优化性能。

保持好奇心，继续编码，你很快就会成为Web开发高手！🚀

如果有问题，随时查阅FastAPI官方社区或教程——学习路上，你并不孤单。