背景痛点：批处理模式在高并发场景下的“三宗罪”
去年双十一，我们第一次把 ChatTTS 接进电商客服的语音机器人，结果凌晨 0 点 30 分直接“炸”了：
- 延迟飙到 3.8 s，用户说完“我要退款”等了快 4 秒才听到回复，体验翻车。
- 内存 5 分钟从 2 GB 涨到 14 GB，Kubernetes 把 Pod 当“内存怪兽”直接 OOMKill，客服线上一度无语音可用。
- 并发一高，GPU 排队，句子像春运买票，后端疯狂 502。
根因并不神秘——ChatTTS 默认的“整句→整段”批处理：
- 要等整句文本全部编码完才开始解码，首包时间 = 整句长度 ÷ 模型吞吐。
- 解码后一次性返回 10 s+ 的 PCM，内存峰值与句长成正比。
- HTTP 短连接，每次重握手，高并发下内核 SYN 队列打满。
一句话，“批”不动实时场景。流式处理成了唯一出路。
技术对比：轮询、SSE 还是 WebSocket？
我们把三种“能一点点吐数据”的方案放在 4 核 8 G 的测试机里，用 100 并发压 30 s 音频，结论如下：
| 方案 | 首包延迟 | 单连接内存 | 断线重连成本 | 代码复杂度 | 结论 |
|---|---|---|---|---|---|
| HTTP 轮询 | 600 ms+ | 低 | 高（反复 3 次握手） | ★ | 直接弃 |
| SSE | 280 ms | 中 | 中（浏览器自动重连） | ★★ | 可用，但仅服务端→客户端单向 |
| WebSocket | 120 ms | 中 | 低（应用层心跳） | ★★★ | 双工、最低延迟，选它 |
SSE 看似轻量，但语音场景需要客户端随时pause/resume，双向通道更省心；再加上 SSE 在 HTTP/2 下仍有队头阻塞风险，最终拍板WebSocket + 二进制帧。
核心实现：搭一条“流式管道”
1. 整体架构
浏览器 ←WebSocket→ 网关(nginx) ←→chatts_stream.py←→ 环形缓冲区 ←→ FFmpeg ←→ 用户耳机
2. 环形缓冲区：让模型与网络解耦
模型按 160 ms 粒度解码，网络 MTU 约 1 400 byte，两边节奏不同步，必须有个“蓄水池”。
自写RingBuffer：
class RingBuffer: """线程安全、固定大小、覆盖写""" def __init__(self, size: int): self._buf = bytearray(size) self._size = size self._head = 0 self._tail = 0 self._lock = asyncio.Lock() async def write(self, data: bytes) -> int: async with self._lock: free = self._size - (self._head - self._tail) if len(data) > free: # 覆盖旧数据 self._tail += len(data) - free for b in data: self._buf[self._head % self._size] = b self._head += 1 return len(data) async def read(self, n: int) -> bytes: async with self._lock: avail = self._head - self._tail if avail <= 0: return b'' real = min(n, avail) start = self._tail % self._size end = (self._tail + real) % self._size self._tail += real if end > start: # 未回卷 return self._buf[start:end] return self._buf[start:] + self._buf[:end] # 回卷拼接时间复杂度：读写均为O(n)，n 为单次数据长度；锁粒度只保护指针，竞争极小。
3. asyncio + WebSocket 生产/消费模型
async def producer(ws: WebSocketServerProtocol, rb: RingBuffer): """把模型吐的 20 ms PCM 块塞进缓冲区""" try: async for pcm in chatts_stream_generate(): # 异步生成器 await rb.write(pcm) except ConnectionClosed: pass finally: await ws.close() async def consumer(ws: WebSocketServerProtocol, rb: RingBuffer): """160 ms 一个包，带序号发给前端""" seq = 0 while True: chunk = await rb.read(16000 * 2 * 0.16) # 16kHz 16bit if not chunk: await asyncio.sleep(0.01) continue await ws.send(struct.pack('>H', seq) + chunk) seq = (seq + 1) % 655364. FFmpeg 实时转码进程管理
前端需要opus@48k节省带宽，但模型输出pcm@16k。
起 FFmpeg 子进程，用stdin/stdout 管道：
proc = await asyncio.create_subprocess_exec( 'ffmpeg', '-f', 's16le', '-ar', '16000', '-ac', '1', '-i', '-', '-f', 'opus', '-application', 'audio', '-', stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE)- 用
asyncio.wait_for(proc.stdout.read(4096), 5)防止死锁。 - 进程异常退出时，前端会收到OPUS EoF 帧，触发重连；后端
try/except后重启 FFmpeg，保证“断流不断管”。
性能优化：把延迟压到 200 ms 以内
1. 压测曲线
在 8 核 A10 GPU 云主机，分别打 50/100/200 并发，30 s 句长：
| 并发 | 首包 P99 | 内存峰值 | GPU 利用率 |
|---|---|---|---|
| 50 | 180 ms | 2.1 GB | 62 % |
| 100 | 195 ms | 2.3 GB | 78 % |
| 200 | 220 ms | 2.5 GB | 83 % |
内存基本平稳，得益于固定大小环形缓冲区；首包延迟随并发线性增长，但斜率 <1 ms/conn，可接受。
2. 令牌桶流控：防止慢客户端拖垮服务端
如果某用户网络卡，TCP 发送窗口积压，服务端的ws.send会阻塞，导致整个事件循环饿死。
实现简单令牌桶：
class TokenBucket: def __init__(self, rate: float, burst: int): self._rate = rate self._tokens = burst self._last = time.monotonic() self._lock = asyncio.Lock() async def consume(self, tokens: int) -> bool: async with self._lock: now = time.monotonic() delta = nownow - self._last self._tokens = min(self._tokens + delta * self._rate, self._rate) self._last = now if self._tokens >= tokens: self._tokens -= tokens return True return False每路连接 80 kB/s 的 opus 码率，桶大小 160 kB，O(1)时间复杂度，对 CPU 几乎无感。
当令牌不足，服务端主动drop 帧并给前端发{"type":"throttle","level":0.8}，前端降采样，保证实时。
避坑指南：上线前必须踩的 3 个坑
WebSocket 保活
浏览器→nginx 默认 60 s 无数据就断。我们在每 20 s 发2 byte ping（opcode 0x9），nginx 配置proxy_timeout 70s;对齐即可。音频分块 vs MTU
160 ms 16-bit 16 kHz 单声道 = 5 120 byte，加上 2 byte 序号，5 122 byte< 1 400 byte，不会 IP 分片；
如果盲目用 320 ms，10 242 byte超过 1 400，导致公网丢包率 2 %→8 %，延迟反而抖动。异常断开后状态同步
用户刷新页面后 WebSocket 重连，服务端要先发当前已合成的文本偏移（字符级），前端从该偏移继续传文本，否则会出现“半句复读”。偏移存在放在Redis TTL 键，5 分钟过期，兼顾断线重连与内存释放。
完整可运行 Demo（核心片段）
# chatts_stream.py import asyncio, json, struct, time from typing import AsyncGenerator from websockets.server import serve, WebSocketServerProtocol async def chatts_stream_generate(text_feed: AsyncGenerator[str, None]) -> AsyncGenerator[bytes, None]: """Mock 20 ms 1024 sample PCM""" async for text in text_feed: for _ in range(len(text) // 4): # 粗略 20 ms/块 yield b'\x00\x01' * 1024 # 1 024 * 2 byte await asyncio.sleep(0.02) async def handler(ws: WebSocketServerProtocol): rb = RingBuffer(2 * 1024 * 1024) # 2 M tok = TokenBucket(80 * 1024, 160 * 1024) async def text_feed(): async for msg in ws: yield json.loads(msg)['text'] producer_task = asyncio.create_task( producer(ws, rb, chatts_stream_generate(text_feed()))) consumer_task = asyncio.create_task(consumer(ws, rb, tok)) try: await asyncio.gather(producer_task, consumer_task) except Exception as exc: print('conn error', exc) finally: producer_task.cancel() consumer_task.cancel() asyncio.run(serve(handler, '0.0.0.0', 8765))延伸思考：TTS + STT 双流合一
目前我们只解决“说”的实时，下一步把STT 也流式化：
- 前端 WebAudio worklet 分 20 ms 发 float32
- 后端用VAD检测尾点，ASR流式输出文本
- 同一 WebSocket 双工通道，文本双向，音频上行下行，“全双工语音对话”就闭环了。
届时令牌桶要做双桶加权，STT 码率小但优先级高；TTS 码率大，优先级低，防止 ASR 被“噎死”。
有兴趣的同学可以先试Mozilla DeepSpeech + 本文环形缓冲区，跑个原型，再逐步替换为自家模型。
把批处理改成流式后，我们线上客服机器人高峰 1.2 w 并发，P99 延迟稳定在 190 ms，内存只比闲时高 18 %。
代码量不大，关键是“让数据动起来”——模型吐一点，网络发一点，内存就像泳池，进多少出多少，永远不会溢。
如果你也在用 ChatTTS 做实时场景，不妨把上面的 150 行脚本跑通，再一点点调优，相信很快就能把延迟压到 200 ms 以内。祝调试顺利，别忘了给 WebSocket 加心跳，不然凌晨 3 点断线会被用户电话叫醒。
