企业官网建设流程全解析

1. 为什么足球数据API偏偏选FastAPI加WebSocket

先说项目背景。我自己维护了一个足球数据聚合服务，对接第三方数据源，把实时比分、进球事件、红黄牌、赔率变化推给下游客户端。最早用的方案是Flask加轮询接口，前端每三秒拉一次，后来接入的客户端一多，MySQL连接池被打满，接口响应从几十毫秒恶化到两秒开外，运营那边天天截图反馈。后来整体重写，选型定为FastAPI加WebSocket，把实时推送完全切成订阅模式，才把服务从泥潭里拉出来。

这个话题最近讨论度一直很高，尤其在高并发IM、实时数据推送这类场景里。我结合这个足球数据API项目，把FastAPI和WebSocket从选型到落地、从单机到横向扩展的整个链路拆开讲一遍，重点说清楚几个关键点：为什么WebSocket是这类场景的对的选择，连接管理器怎么写才扛得住上万连接，跨Worker广播怎么用Redis中转，以及我在压测和生产环境里真实踩过的坑。适合正在做实时推送服务、或者准备把现有轮询接口改成推送模式的人参考。

1.1 实时推送场景的三个硬需求

足球数据API有别于普通业务API，它的流量特征非常鲜明：

第一是高频率小消息。一场比赛90分钟，进球、射门、角球、换人、VAR复核，每秒可能产生多条事件，每条消息体量很小，几十到几百字节。如果客户端频繁发起HTTP请求，服务端和客户端都在为连接握手、报文头、解析逻辑付出大量固定开销，90%的资源都浪费在“没有新数据”的空轮询上。

第二是强时效性。进球消息晚推送三秒，用户刷到的比分就比电视慢半拍，这在体育数据产品里是不可接受的。WebSocket服务端可以主动推送，消息延迟可以压到毫秒级，而不需要等客户端下一次轮询。

第三是连接长驻。前端页面、移动端App、甚至一些大屏展示，往往在比赛期间数小时保持在线。这种长连接场景天然适配WebSocket，HTTP轮询反而要做大量的连接反复建立和销毁。

这三点叠加，WebSocket几乎是唯一合理的技术选型。短连接轮询在消息频率低时还能勉强工作，一旦赛事密集开启——比如五大联赛同时开打，几千个客户端同时盯数据——并发和带宽的双重压力会直接把轮询方案压垮。

1.2 FastAPI在其中的位置：不只是“快”

FastAPI在这个项目里承担的是三层职责：REST接口层提供历史数据查询、赛事列表等一次性数据获取；WebSocket端点负责实时事件流推送；后台任务负责从数据源服务拉取数据并分发到订阅通道。

选择FastAPI而没有选Flask或Django Channels，核心原因有几个：

原生异步支持。FastAPI基于Starlette，整个请求生命周期都是async/await模型。WebSocket是长连接，处理过程中大量时间在等待I/O——等待客户端消息、等待队列数据、等待Redis返回。这些等待在异步模型里几乎不消耗线程资源。同样是维持一万个连接，用线程模型的Flask需要上万条线程，每条线程默认8MB栈空间，光内存就是几十GB起步；FastAPI的异步模型在单进程内可以用任务（Task）来托管这些连接，内存占用低一个数量级。

类型校验和自动文档。足球数据API的消息结构复杂，一个进球事件可能包含球员ID、球队ID、比赛分钟、得分、是否乌龙、是否点球等十几个字段。FastAPI的Pydantic声明式校验可以保证消息格式不跑偏，同时自动生成OpenAPI文档，前端联调时直接看文档就能对接。

生态顺滑。Uvicorn作为ASGI服务器天然和WebSocket协议兼容；Redis异步客户端、HTTPX异步请求库都能无缝衔接，不阻塞事件循环。

当然，FastAPI并非全无短板。如果项目里有大量CPU密集型计算，比如实时赔率模型推算，异步服务照样会被阻塞，需要把这类任务丢到独立进程里跑。我在这个项目里就把赔率计算单独拆了一个服务，API层只做转发和推送，避免“一粒老鼠屎坏了一锅粥”。

2. 项目骨架与核心目录设计：先把数据流跑通再谈并发

很多初学者拿到一个FastAPI项目，第一件事就是往main.py里堆代码。路由写一堆、WebSocket写一堆、后台任务也塞进去，等到要加功能时一个文件几百行，改一处崩三处。高并发项目更要重视目录结构，因为并发问题往往不是单个函数写得不好，而是模块间的耦合把资源竞争放大了。

2.1 目录结构：官方风格和实战习惯的折中

这是我的项目目录结构，参考了FastAPI官方推荐的拆分方式，同时按照实时推送场景做了一些调整：

football_api/ ├── app/ │ ├── main.py # FastAPI实例、路由注册、启动事件 │ ├── config.py # 全局配置：Redis地址、心跳时间、连接上限 │ ├── api/ │ │ ├── rest.py # REST接口：赛事列表、历史比分查询 │ │ └── ws.py # WebSocket端点和连接管理逻辑 │ ├── core/ │ │ ├── connection.py # ConnectionManager：连接注册/注销/广播 │ │ ├── auth.py # Token校验、握手鉴权 │ │ └── logger.py # 结构化日志配置 │ ├── services/ │ │ ├── data_ingest.py # 从第三方数据源拉取比赛事件 │ │ ├── event_bus.py # Redis发布订阅封装 │ │ └── match_service.py # 赛事数据业务逻辑 │ ├── models/ │ │ ├── match.py # 赛事、事件、赔率的Pydantic模型 │ │ └── ws_message.py # WebSocket消息协议模型 │ └── utils/ │ ├── redis_client.py # Redis连接池 │ └── helpers.py # 通用工具函数 ├── tests/ ├── deploy/ │ ├── nginx.conf │ └── docker-compose.yml ├── .env └── requirements.txt

分层逻辑很清晰：API层只负责握手、鉴权、收发消息，不碰业务逻辑；Service层处理数据源对接和业务规则；Core层是基础设施，连接管理器和事件总线这些所有模块共享的东西放这里。好处是，当你要排查WebSocket连接异常时，不需要去data_ingest.py里翻代码——各模块的职责边界一目了然。

2.2 数据源层与API层的职责拆分

这个项目里最容易搞乱的就是数据源拉取和API推送的关系。最初我犯过一个错误：在WebSocket接收消息的Handler里直接去请求上游数据源。结果上游服务偶尔抖动，一个Handle卡了五秒，整个事件循环被堵住，所有连接都跟着遭殃。

正确做法是彻底分离：

数据源层（data_ingest.py）单独跑一个后台异步任务，从第三方服务拉取比赛事件，做格式转换、脏数据清洗，然后发布到Redis频道。API层（ws.py）只负责维护客户端连接，从Redis订阅频道拿到消息后广播给对应的客户端。

这样设计之后，就算数据源响应慢，影响的也只是拉取任务本身，推送链路完全不受影响。而如果某个客户端消费速度慢，也只是它自己的连接被阻塞，不会拖累其他连接。

3. WebSocket连接管理：从单连接到万级并发的关键设计

WebSocket比HTTP复杂的地方在于连接是有状态的。HTTP请求来了就处理、处理完就断开，服务器不需要记住客户端；WebSocket连接建立后要一直维护，客户端可能随时上线、掉线、重连，服务端还必须主动给它推消息。所以整个项目的核心就是连接管理器的设计。

3.1 连接注册表与连接生命周期的完整代码

先给一个可以直接用的ConnectionManager。这个版本我是在生产环境打磨过的，支持按频道维度分组广播：

import asyncio import time from collections import defaultdict from typing import Dict, Set, Optional from fastapi import WebSocket class ConnectionManager: def __init__(self): # 所有活跃连接，key为client_id self.active_connections: Dict[str, WebSocket] = {} # 连接所属的频道订阅关系，channel -> set of client_id self.channel_members: Dict[str, Set[str]] = defaultdict(set) # 记录每个连接最近一次心跳时间 self.last_heartbeat: Dict[str, float] = {} # 记录每个连接订阅的频道，用于断开时清理 self.client_channels: Dict[str, Set[str]] = defaultdict(set) self._lock = asyncio.Lock() async def connect(self, client_id: str, websocket: WebSocket) -> None: await websocket.accept() async with self._lock: self.active_connections[client_id] = websocket self.last_heartbeat[client_id] = time.time() async def disconnect(self, client_id: str) -> None: async with self._lock: self.active_connections.pop(client_id, None) self.last_heartbeat.pop(client_id, None) for channel in list(self.client_channels.get(client_id, set())): self.channel_members.get(channel, set()).discard(client_id) if not self.channel_members.get(channel): self.channel_members.pop(channel, None) self.client_channels.pop(client_id, None) async def subscribe(self, client_id: str, channel: str) -> None: async with self._lock: self.channel_members[channel].add(client_id) self.client_channels[client_id].add(channel) async def unsubscribe(self, client_id: str, channel: str) -> None: async with self._lock: self.channel_members.get(channel, set()).discard(client_id) self.client_channels.get(client_id, set()).discard(channel) async def send_to_client(self, client_id: str, message: dict) -> bool: ws = self.active_connections.get(client_id) if not ws: return False try: await ws.send_json(message) return True except Exception: # 发送失败说明连接已经断开，触发清理 await self.disconnect(client_id) return False async def broadcast_to_channel(self, channel: str, message: dict) -> int: """广播到指定频道，返回成功发送的连接数""" async with self._lock: member_ids = list(self.channel_members.get(channel, set())) sent = 0 for client_id in member_ids: if await self.send_to_client(client_id, message): sent += 1 return sent async def broadcast_all(self, message: dict) -> int: """全局广播，返回成功发送的连接数""" async with self._lock: client_ids = list(self.active_connections.keys()) sent = 0 for client_id in client_ids: if await self.send_to_client(client_id, message): sent += 1 return sent

这里有个关键设计：连接字典、频道订阅关系、心跳时间都用同一个asyncio.Lock保护。WebSocket的消息处理是并发执行的，多个协程可能同时修改连接状态——比如一个协程在广播，另一个协程在处理客户端断开，如果不加锁，字典在遍历时被修改，直接抛RuntimeError。锁的粒度不需要太小，因为这里面的操作都是内存级别的，即使全局加锁，单机每秒也能处理数万次操作，不会成为瓶颈。

3.2 心跳机制的正确实现：防止僵尸连接吃满CPU

WebSocket本身没有内置超时断连机制，TCP层可能很久都感知不到对端已经消失。如果客户端断网但没有正常关闭连接，服务器端的连接对象会一直存在，这就是“僵尸连接”。连接少时无所谓，连接多了之后，每次全局广播都要遍历所有僵尸连接，对已经失效的socket调用send_json，抛异常后再清理——每一次失败的发送都白白浪费资源。

正确做法是心跳机制。业界标准做法是用Ping/Pong帧，客户端定时收到服务端Ping后回Pong，服务端通过Pong判断连接活跃。我在这个项目里用的是业务层心跳：

HEARTBEAT_INTERVAL = 30 # 每30秒检查一次 HEARTBEAT_TIMEOUT = 90 # 超过90秒没有心跳就判定死亡 async def heartbeat_loop(manager: ConnectionManager): while True: await asyncio.sleep(HEARTBEAT_INTERVAL) now = time.time() dead_clients = [] async with manager._lock: for client_id, last_time in manager.last_heartbeat.items(): if now - last_time > HEARTBEAT_TIMEOUT: dead_clients.append(client_id) for client_id in dead_clients: await manager.disconnect(client_id) # 通知业务侧客户端已被清理 logger.warning(f"client {client_id} heartbeat timeout, disconnected")

客户端收到业务消息时，会在消息里带上时间戳，客户端定时发送{"type": "ping"}，服务端在接收消息的Handler里更新last_heartbeat。注意心跳检查和连接清理不能混在一个锁里——检查锁只用来读取快照，清理是逐连接异步执行，避免清理慢连接时阻塞其他所有连接的操作。

3.3 鉴权握手：在WebSocket入口拦住非法连接

WebSocket建立连接时发的是HTTP Upgrade请求，所以可以复用HTTP的鉴权方式。推荐在URL参数里带短期Token，配合签名校验。完整的握手流程如下：

from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Query router = APIRouter() @router.websocket("/ws") async def websocket_endpoint( websocket: WebSocket, token: str = Query(...), client_id: str = Query(...), ): # 第一步：校验token if not validate_token(token): await websocket.close(code=4401) return # 第二步：检查是否重复连接 manager = get_connection_manager() if client_id in manager.active_connections: # 新连接挤掉旧连接，这是体育数据场景的常见需求 await manager.disconnect(client_id) # 第三步：注册连接 await manager.connect(client_id, websocket) # 第四步：回复握手成功消息 await websocket.send_json({"type": "connected", "client_id": client_id}) # 第五步：进入消息接收循环 try: while True: data = await websocket.receive_json() # 更新心跳 if data.get("type") == "ping": async with manager._lock: manager.last_heartbeat[client_id] = time.time() continue # 处理订阅请求 if data.get("type") == "subscribe": channel = data.get("channel") await manager.subscribe(client_id, channel) await websocket.send_json({ "type": "subscribed", "channel": channel, }) elif data.get("type") == "unsubscribe": channel = data.get("channel") await manager.unsubscribe(client_id, channel) except WebSocketDisconnect: await manager.disconnect(client_id)

Token校验失败时用4401关闭连接，而不是直接断掉TCP，这样客户端能收到明确的错误码来做区分处理。重复连接直接踢掉旧的，这在移动端场景里很重要——用户断网重连后，网络层可能拿同一个client_id建立新连接，不处理的话服务器上会有两条连接都认为自己是同一个人，推送时可能发到旧连接上，新连接反而收不到。

4. 订阅协议与广播策略：让每个客户端只收到该看的数据

如果只有一个全局广播频道，把所有比赛数据推给所有客户端，实现最简单，但并发一上来就完蛋。一个比赛日可能有几十场比赛同时进行，每个客户端通常只关心其中一两场。全局广播意味着所有消息推给所有人，带宽和数据量放大几十倍，客户端还得自己做过滤。所以必须设计频道化的订阅协议。

4.1 按赛事与按球队的频道设计

足球数据的频道设计我建议分成两级：

赛事频道：match.{match_id}，推送该场比赛的所有实时事件，包括进球、射门、角球、换人、比分变化、比赛状态变化。这是最常用的订阅维度。

球队频道：team.{team_id}，推送该球队相关的所有比赛新闻和事件。球迷关注一支球队，可能同时关心它今天在联赛和杯赛的表现，球队频道可以聚合多场比赛的消息。

联赛频道：league.{league_id}，推送给运营方或者做数据分析的客户端，需要同时接收整个联赛所有比赛的比分变化。

客户端可以在一个连接里同时订阅多个频道，比如订阅match.1001和team.789，这样一场进球事件会同时推送到两个频道，客户端会收到两条内容相似但场景不同的消息。实现上不复杂——连接管理器里每个连接维护一个频道集合，广播时按频道过滤连接即可。

频道命名要规整，解析和权限控制都方便。我在代码里用split(".")来解析频道名，第一段是资源类型，第二段是资源ID。订阅时校验客户端是否有权限订阅这个频道，比如免费用户只能订赛事频道，付费用户才能订赔率频道。

4.2 Redis发布订阅作为跨Worker广播的“中转站”

单进程能撑住的连接数有限，Uvicorn多Worker是必然选择。但多Worker带来了一个经典问题：客户端A连在Worker 1，客户端B连在Worker 2，当数据源拉取了一条进球消息，该广播给谁？每个Worker只知道自己的连接，如果Worker 1把消息只发给自己维护的连接，Worker 2的客户端就收不到。

解决方案就是Redis发布订阅（Pub/Sub）。所有Worker都订阅同一个Redis频道，数据源把这个频道发布消息，所有Worker收到后再转发给自己维护的连接。这就是“一个生产者，多个消费者，然后再分发到各自的连接池”的模型。

# event_bus.py import json import redis.asyncio as aioredis CHANNEL_ALL = "football:events" class EventBus: def __init__(self, redis_url: str): self.redis = aioredis.from_url( redis_url, decode_responses=True, max_connections=50, ) self.pubsub = None async def publish(self, channel: str, message: dict) -> None: payload = json.dumps(message, ensure_ascii=False) # 统一推到总频道，所有Worker都能收到 await self.redis.publish(f"{CHANNEL_ALL}:{channel}", payload) async def subscribe_loop(self, callback): """在后台任务中启动订阅循环""" self.pubsub = self.redis.pubsub() await self.pubsub.subscribe(f"{CHANNEL_ALL}:*") async for message in self.pubsub.listen(): if message["type"] != "message": continue channel = message["channel"] data = json.loads(message["data"]) await callback(channel, data)

Redis Pub/Sub的subscribe_loop需要放在每个Worker的启动事件里。其实更好的做法是启动一个专门的订阅任务，把回调函数绑定到ConnectionManager的广播方法：

@app.on_event("startup") async def startup(): event_bus = get_event_bus() manager = get_connection_manager() async def on_event(channel: str, data: dict): # channel形如 football:events:match.1001 # 提取真正的业务频道名，只广播给对应订阅者 business_channel = channel.split(":", 2)[-1] await manager.broadcast_to_channel(business_channel, data) asyncio.create_task(event_bus.subscribe_loop(on_event))

注意：Redis Pub/Sub是“即发即弃”的。如果某个Worker在消息发布时还没有建立订阅连接，或者正在重启，它会漏掉这条消息。对于足球比分这种状态类消息，漏掉一条爆炸性消息影响很大。我在项目里加了兜底方案——每30秒从MySQL拉一次全量比赛状态做快照，推送给新订阅的客户端，保证订阅时客户端至少拿到当前正确状态，后续增量事件通过Redis订阅补齐。这就是“快照+增量”的消息模型，也是金融行情推送的常用思路。

4.3 消息格式设计与客户端容错

消息格式统一用JSON，结构固定如下：

{ "type": "match.event", "match_id": 1001, "event_id": "evt_8890123", "timestamp": 1712345678, "data": { "minute": 67, "type": "goal", "team": "home", "player": "Erling Haaland", "score_home": 2, "score_away": 1 } }

type字段决定了客户端如何处理这条消息。我在服务端用Pydantic模型约束消息格式，写出去之前先validate一遍，避免脏数据直接推给客户端：

from pydantic import BaseModel, Field class MatchEventMessage(BaseModel): type: str = Field(..., pattern="^match\.") match_id: int event_id: str timestamp: int data: dict

客户端收到消息后，根据type分发给对应处理函数。服务端连续推送时客户端可能来不及处理，所以需要考虑背压。WebSocket send_json本身是异步的，如果客户端消费速度慢，socket的发送缓冲区会越来越大。极端情况内存暴涨。我建议在send_to_client里增加一个简单的流量控制：如果某个连接待发送的消息数量超过阈值，主动断开这个客户端，让它重连。

5. 高并发实战中的性能调优与压测记录

基础功能跑通之后，我开始做压测和调优。这里分享我实测的数据和调优方向。

5.1 Uvicorn参数与进程模型选择

Uvicorn支持两种进程模型：单进程多 Worker 和单进程单事件循环。对于WebSocket场景，多Worker模式是必须的，因为单进程的CPU核心利用率有限，而且事件循环处理大量连接时，单个进程的fd数量、内存都容易成为瓶颈。

我实测的启动参数：

uvicorn app.main:app \ --host 0.0.0.0 \ --port 8000 \ --workers 4 \ --ws max-size=1048576 \ --limit-concurrency 20000 \ --limit-max-requests 0 \ --backlog 2048

--ws max-size控制WebSocket单条消息最大大小，我设成1MB，防止客户端发送超大消息刷爆内存。--limit-concurrency是Uvicorn层面的并发连接上限，设成2万，超过就拒绝新连接，保护服务不被打垮。--backlog是TCP层的accept队列长度，并发连接涌入时很重要——队列满了内核直接丢连接，客户端表现为连接被重置。

但多Worker也要注意：Uvicorn多个Worker绑定同一个端口，由内核做负载均衡。对于WebSocket长连接，内核会把连接分发到不同的Worker，每个Worker独立维护自己的连接集合。这个和上面提到的Redis跨Worker广播正好配合。

5.2 实际压测结果与瓶颈分析

我的压测环境是两台4核8G的云主机，一台跑服务，一台跑压测脚本。压测脚本用Python的websockets库，模拟5000个客户端并发连接，每个客户端订阅一个随机赛事的频道。数据源模拟每秒钟产生500条比赛事件。

第一次压测结果很扎心：连接数到2000左右，新连接就开始大量失败。排查发现瓶颈不在应用层，在系统层：

文件描述符上限。默认ulimit -n是1024，一个WebSocket连接至少占用一个fd，加上Redis连接、日志文件，2000连接就已经顶到系统限制。修改：

ulimit -n 65535

以及/etc/security/limits.conf里把nofile调大，重启后生效。

TCP内核参数。高并发连接会触发大量TIME_WAIT和连接队列问题，我把以下参数写进了/etc/sysctl.conf：

net.ipv4.tcp_fin_timeout = 10 net.ipv4.tcp_tw_reuse = 1 net.ipv4.ip_local_port_range = 1024 65535 net.core.somaxconn = 2048

改完之后，重新压测，5000连接稳定建立，消息推送延迟P99在80ms以内，整体吞吐能满足需求。第二次优化把Redis的maxmemory和databases做了调优，把pub/sub连接数和主服务连接数分开管理，避免Redis成为瓶颈。

再往后压测到8000连接时，发现CPU使用率开始飙升，定位后发现定时心跳任务每30秒遍历一次所有连接，全是活跃连接还好，有大量半开连接时每次超时都要等send超时返回。把心跳超时从90秒改成45秒，并且把心跳遍历改为只遍历last_heartbeat字典而不是遍历连接对象本身，删除连接时马上清理字典，CPU占用降回来不少。

5.3 数据库层的减压方案

WebSocket推送链路中，数据库压力来自两个方向：REST接口的历史数据查询，以及数据源拉取时的事件落库。足球数据写入频率高，如果每来一个事件就INSERT一次，压力很大。

我用的是批量写策略：数据源拉取的事件先放进内存队列里，每5秒或者累计满500条再批量写库。这样写入频率从每秒几十次降为每5秒一次，MySQL的压力小了很多。查询侧加了Redis缓存热点数据，比如当前比赛日的赛事列表缓存30秒，历史比赛比分缓存5分钟。

6. 生产环境踩坑实录：从连接数上不去到内存暴涨的排查链路

这部分是项目上线后真实遇到的问题，排查过程比结果更有价值。

6.1 现象一：超过一千连接后新连接被拒

上线当天就炸了。运营那边反馈客户端连不上，日志里大量Connection closed。先看Uvicorn日志，发现没有异常堆栈，再查系统参数，发现max_user_instances、文件描述符都正常。最后用ss -s查看socket统计，发现timesec和established的数值很高，再查/proc/sys/net/ipv4/tcp_max_syn_backlog默认只有256，高并发握手时SYN队列满了，内核直接丢弃新连接。调大tcp_max_syn_backlog和somaxconn之后恢复正常。

这个问题的本质是：虽然WebSocket连接是长连接，但建立连接那一刻仍然要经过完整的TCP三次握手。如果握手队列太短，连接风暴一来就把新连请求全丢了，排查时容易直接被“应用层没问题”误导。

6.2 现象二：多Worker启动后客户端收不到广播

上线第二天，新功能上线后，部分客户端反馈订阅了频道但收不到消息。单Worker测试时一切正常，多Worker部署后必现。我花了一个小时排查，最后在Redis的Pub/Sub客户端源码里发现问题：subcribe_loop里用了pattern subscribe（PSUBSCRIBE），但listen()方法默认只能收到精确匹配的消息。踩在模式订阅和精确订阅的坑上。改用pmsg字段读取实际频道名后，问题解决。

这不是FastAPI的问题，是Redis异步客户端API使用细节。教训是：多Worker模式下，任何跨进程的消息传递都要用真实消息系统，而且要对中间件的API非常熟。

6.3 现象三：内存持续上涨，三个小时后重启

上线跑了一周，某天凌晨值班报警说内存占用跑到90%。登进去一看，Python进程RSS已经到3.5G，持续上涨没有回落趋势。用tracemalloc和objgraph排查，发现是WebSocket连接的接收循环里，websocket.receive_json()返回的dict没有释放。原因是我在消息处理分支里，订阅和退订操作频繁修改ConnectionManager的client_channels和channel_members，但断开连接时只清理了active_connections和last_heartbeat，忘了清理client_channels，导致每次客户端正常断开后，订阅关系残留在内存里，越来越肥。

修复方案就是在disconnect方法里，把client_channels和channel_members的引用也全部清理干净。除了代码修复，还加了内存自动告警和定时重启的应急兜底。

回头总结这个问题的排查思路，最重要的一步是：当怀疑内存泄漏时，一定要统计“哪些对象在持续增长”，而不是靠猜。用objgraph.show_growth()打印出增长最多的对象类型，看到dict和set暴涨，再去查哪些代码路径在无限制地往dict和set里添加东西，一击即中。

写在最后的一点实战心得

整个项目从选型到稳定运行，我最大的体会是：FastAPI加WebSocket的组合，在实时数据推送场景里非常趁手，但框架只帮你解决了“建立连接、收发消息”这部分，真正决定高并发成败的是连接管理、心跳策略、广播通道、跨进程消息这几个层面的设计。足球数据这个业务场景很有代表性——消息频率高、时效性强、连接数大、客户端网络环境复杂，把这些问题的解法沉淀下来，以后做行情推送、在线IM、协作编辑这类场景，思路完全可以复用。

如果你也想做同类项目，我最后建议三件事：一是从第一天就画好连接生命周期的状态图，明确每个状态下的清理逻辑；二是压测一定要打满系统参数，否则你会把系统层瓶颈误判成代码问题；三是务必给WebSocket服务接上完善的指标监控，连接数、消息吞吐、心跳超时数、内存趋势这些指标一个都不能少。数据驱动的排查方式，比靠感觉定位问题高效得多。