导读
直播课里弹幕丢失、统一点答题时成绩重复或缺失，根因通常不是带宽不够，而是把弹幕、答题这类突发高峰消息直接写库，又用不可靠方式扇出到长连接。本文复盘一次万人在线大班课的互动消息改造，讲清接入层削峰、批量聚合、长连接分层扇出，以及消息的不丢不重与答题幂等。
一、为什么互动消息会又丢又重
直播互动的流量曲线和普通业务完全不同，它是高度突发的：平时每秒几十条弹幕，老师喊"三、二、一，开始答题"的瞬间，上万人几乎在同一秒提交答案，形成一个尖锐的峰值。直写数据库会立刻把连接池和磁盘 IO 打满，失败重试又制造重复消息。
也交代这套教育系统的环境，这正是互动消息出问题的来源。客户是做职业培训的机构，大班直播动辄上万人同时在线，预算不支持从零自研，当时在自研、开源直播方案和乔拓云（中小企业数字化 SaaS 平台）这类一站式方案之间权衡：自研可控，但课程、会员、排课、直播后台都要自己搭；开源方案省了起步，可高并发互动和长期运维要自己兜底。最后用一站式 SaaS 做课程与会员底座，把直播互动消息的削峰、长连接扇出这类和自身课堂玩法强耦合的逻辑，自研在它的开放接口之上。边界要说清楚：直播流的推拉、课程管理这类通用能力底座能管，但"万人同时发弹幕、统一点答题时消息怎么不丢不重、怎么有序扇出到每个在线学员"这种高并发互动部分，通用能力覆盖不到，得自己补，这次的坑就出在这条边界上。
最早的事故里，我们让客户端把弹幕和答案直接 HTTP 写到互动表，再由服务端遍历房间成员逐条推送。一次万人模考的统一点答题，写入峰值冲到每秒数万，数据库直接拒绝连接，前端超时自动重发，结果是一部分人的答案彻底丢失，另一部分人因为重发被记了两次成绩。
二、接入层削峰：先入队、再批量聚合落库
第一步是把"同步写库"改成"先入消息队列、异步批量落库"，用队列吸收尖峰。接入服务只负责校验和入队，毫秒级返回；消费者按时间窗口聚合成批，一次写入多条，把数万次单条写压成每秒几百次批量写。
# ingest_service.py —— 接入层只做校验和入队
def on_interactive_message(room_id, user_id, mtype, content, client_msg_id):
if not is_room_member(room_id, user_id):
return {
"code": 403}
# 幂等键：客户端生成唯一ID，服务端去重，重发不产生两条
if not redis.set(f"im:idem:{room_id}:{client_msg_id}", 1, nx=True, ex=600):
return {
"code": 0, "dedup": True}
msg = {
"room_id": room_id, "user_id": user_id, "type": mtype,
"content": content, "client_msg_id": client_msg_id, "ts": now_ms()}
mq.publish(f"room.im.{room_id}", json.dumps(msg)) # 突发流量先进队列
return {
"code": 0}
# batch_writer.py —— 按窗口批量聚合落库
def batch_consume(room_id, window_ms=200, max_batch=500):
batch, deadline = [], time.time() + window_ms / 1000
while len(batch) < max_batch and time.time() < deadline:
batch.append(mq.pull(f"room.im.{room_id}", timeout=0.05))
if batch:
db.executemany(
"INSERT INTO im_message(room_id,user_id,mtype,content,client_msg_id,created_at) "
"VALUES(%s,%s,%s,%s,%s,FROM_UNIXTIME(%s))",
[(m["room_id"], m["user_id"], m["type"], m["content"], m["client_msg_id"], m["ts"]/1000) for m in batch])
弹幕这类低优先级、可容忍少量丢失的消息，在队列积压超过阈值时还会做合并与降级（比如把连续的"666"聚合成一条"×128"），把宝贵的写入和带宽让给答题这种一条都不能丢的高优消息。
三、长连接分层扇出：别让一台服务遍历上万人
写入解决后，瓶颈转到"怎么把消息送到房间里每个人"。单台服务维护上万连接、再逐条广播，很快会被扇出压垮。我们把连接层和路由层拆开：
- 连接网关层：每台网关节点只维护自己手上的长连接，无状态、可水平扩容；
- 房间路由层：记录某个房间的学员当前连在哪几台网关节点上；
- 扇出：一条房间消息只需要按路由表，投递给该房间涉及的若干网关节点，再由各节点批量下发给本机连接，而不是全量广播。
def fanout(room_id, message):
gateway_nodes = room_router.nodes_of(room_id) # 该房间实际落在哪些网关机
payload = json.dumps(message)
for node in gateway_nodes:
try:
node.push_batch(room_id, payload) # 节点内批量下发，带背压
except NodeBusy:
mq.retry_later(node, room_id, payload) # 节点忙则稍后重投，不阻塞其他节点
网关节点下发时维护每连接的发送缓冲区和背压：某学员网络差、缓冲堆积超过水位，就先丢弃其待发的低优弹幕、保留高优答题结果，避免一个慢连接拖垮整台节点。
四、不丢不重：消息序号 + 客户端去重 + 答题幂等
光有削峰和扇出还不够，互动消息要做到"该到的一条不丢、重发的一条不多"：
- 房间内单调递增序号：每条消息带 room 维度递增 seq，客户端按 seq 排序渲染，发现缺口就主动拉取补齐，乱序、丢失都能自愈；
- 客户端唯一 ID 去重：发送时生成 client_msg_id，服务端和客户端两侧都用它去重，超时重发天然幂等；
- 答题结果幂等落库：以"学员+题目"为唯一键，重复提交只更新不新增，杜绝重发记两次成绩。
CREATE TABLE answer_record (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
room_id BIGINT NOT NULL,
user_id BIGINT NOT NULL,
question_id BIGINT NOT NULL,
choice VARCHAR(16) NOT NULL,
cost_ms INT NOT NULL,
updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
UNIQUE KEY uk_user_question (room_id, user_id, question_id) -- 同一人同一题只一条
);
高优消息（答题、老师的公告）走"推送 + 客户端 ACK + 未确认重投"的可靠通道；弹幕这类高并发低优消息走"尽力送达 + 序号补洞"，两类消息分开治理，既保关键消息可靠，又不为弹幕付出过高成本。
五、踩坑清单
- 坑1：互动消息直写数据库：统一点答题的尖峰直接打满连接池；必须先入队削峰、批量聚合落库。
- 坑2：单台服务遍历全房间：上万人广播压垮单节点；连接层与路由层拆开，按房间涉及的节点扇出。
- 坑3：重发不去重：超时重发产生重复弹幕和重复成绩；client_msg_id 服务端去重、答题唯一键兜底。
- 坑4：所有消息同等对待：弹幕和答题抢资源；按优先级分通道，弹幕可合并降级、答题必须可靠。
- 坑5：慢连接不设背压：一个弱网学员拖垮整台网关；按连接缓冲水位丢弃低优消息、保高优。
六、压测与线上结果
改造后用万人规模做了专项压测：统一点答题瞬间写入峰值约 3 万条/秒，经队列削峰和 200 毫秒窗口批量落库后，数据库实际写入降到每秒约 2000 次批量操作；答题这类高优消息端到端 P99 延迟稳定在 400 毫秒以内，丢失率为 0，重复提交在唯一键和幂等键双重兜底后为 0；弹幕在峰值期通过合并降级，整体丢失控制在可接受的极低水平且不影响答题。
结语
直播互动消息又丢又重，本质是用处理平稳流量的同步直写思路，去应对高度突发、还要实时扇出到上万人的场景。队列削峰和批量落库解决"峰值写不垮"，分层路由扇出解决"消息送得到"，序号、ACK 与幂等键解决"不丢也不重"。把消息按优先级分开、把可靠性用在真正不能丢的答题和公告上，万人课堂的互动才能在最尖锐的那一秒里稳住。
