文章目录一、整体架构一条单向数据流二、事件模型三种事件一个基类三、接入层JDK 自带的 WebSocket 客户端断线重连指数退避别死循环硬怼四、解析层报文转事件协议差异全关在这一层五、分发层有界队列 背压丢弃全文最关键的一段六、组装与实测输出七、从 Demo 到生产还差这几步直播间数据采集的教程多数停在「连上 WebSocket 就能收弹幕」。真到高人气直播间你会撞上一个更麻烦的问题消息来得比处理得快。弹幕高峰期一秒钟几千条解析稍慢、落库稍慢消息就开始积压内存涨、延迟飙最后客户端被平台断开连接。所以一个能扛量的小框架缺的不是「连上」是三层结构接入层WebSocket 断线重连、解析层报文 → 事件、分发层事件总线 背压控制。这篇把三层全部写出来核心链路已在本机编译运行实测输出贴在文中。先说清楚合规边界这段不是免责套话采集只针对公开直播数据公屏弹幕、礼物、点赞数必须遵守各平台的开发者协议与频率限制不得绕过鉴权、不得采集用户隐私、不得用于刷量等灰产用途。各平台均有官方开放接口优先用官方接口本文的框架是为了让你理解消息怎么流、系统怎么扛。环境要求JDK 11WebSocket 客户端用的是 JDK 自带的java.net.http.HttpClient零第三方依赖不用引 Java-WebSocket 那套库。一、整体架构一条单向数据流平台 WebSocket 服务器 │ 原始报文JSON / 二进制协议 ▼ ┌─────────────────┐ │ 接入层 WsClient │ 连接、心跳、断线重连指数退避 └────────┬────────┘ ▼ ┌─────────────────┐ │ 解析层 Parser │ 报文 → DanmakuEvent / GiftEvent / LikeEvent └────────┬────────┘ ▼ ┌─────────────────┐ 有界队列1万条 │ 分发层 EventBus │ ───────────────► 单线程按类型分发给监听器 └─────────────────┘ 队列满丢最旧 → 保证 IO 线程永不阻塞 ▼ 落地控制台 / 落库 / 统计 / 推送设计原则就两条单向流动层与层之间只传事件。哪一层出问题换掉哪一层就行。宁可丢数据不能阻塞接收。直播数据的价值随时间快速衰减3 秒前的弹幕和现在的弹幕不是一回事阻塞反而会让连接被服务端判死。二、事件模型三种事件一个基类/** 所有直播事件的基类统一携带时间戳 */abstractclassLiveEvent{finallongtsSystem.currentTimeMillis();}classDanmakuEventextendsLiveEvent{Stringuser,text;DanmakuEvent(Stringuser,Stringtext){this.useruser;this.texttext;}}classGiftEventextendsLiveEvent{Stringuser,gift;intcount;GiftEvent(Stringuser,Stringgift,intcount){this.useruser;this.giftgift;this.countcount;}}classLikeEventextendsLiveEvent{Stringuser;intcount;LikeEvent(Stringuser,intcount){this.useruser;this.countcount;}}为什么先定义事件而不是直接在消息回调里 if/else因为下游要按事件类型订阅比如统计模块只要礼物事件有了类型体系分发层才能「按类找监听器」。三、接入层JDK 自带的 WebSocket 客户端JDK 11 起HttpClient内置了 WebSocket 支持不需要任何依赖importjava.net.URI;importjava.net.http.*;importjava.time.Duration;HttpClientclientHttpClient.newBuilder().connectTimeout(Duration.ofSeconds(10)).build();WebSocketwsclient.newWebSocketBuilder().buildAsync(URI.create(wss://你的平台弹幕网关),newWebSocket.Listener(){// 服务端推来的每条消息都会进 onTextOverridepublicjava.util.concurrent.CompletionStage?onText(WebSocketwebSocket,CharSequencedata,booleanlast){Stringframedata.toString();LiveEventeventparser.parse(frame);// 交给解析层if(event!null){bus.publish(event);// 交给分发层}webSocket.request(1);// 请求下一条注意这句不能少returnnull;}OverridepublicvoidonError(WebSocketwebSocket,Throwableerror){reconnect();// 出错后走重连}}).join();三个要点onText里必须调webSocket.request(1)否则默认只收一条消息就停了。这是这个 API 最反直觉的地方忘了调的表现是「连接正常但再也不收消息」。心跳要主动发。弹幕网关通常要求客户端每 30~60 秒发一帧心跳各平台帧格式不同超时不发会被踢。用ScheduledExecutorService定时ws.sendText(心跳帧, true)。真实平台的报文很多是二进制协议甚至压缩过的比如斗鱼是自定义文本协议、B 站是 zlib/brotli 压缩的 protobuf 队列onText换成onBinary解压和拆包逻辑放解析层。接入层的结构不变。断线重连指数退避别死循环硬怼/** 指数退避连续失败时重连间隔 1s → 2s → 4s → 8s ... 上限 60s */classReconnectPolicy{privatelongdelay1;longnext(){longddelay;delayMath.min(delay*2,60);returnd;}voidreset(){delay1;}// 连接成功后重置}重连循环的骨架voidconnectLoop(){while(running){try{connect();// 建连 注册监听reconnectPolicy.reset();// 成功了归零Thread.sleep(30_000);// 心跳维持简化示意}catch(Exceptione){longwaitreconnectPolicy.next();System.err.println(连接失败waits 后重试: e.getMessage());try{Thread.sleep(wait*1000);}catch(InterruptedExceptionie){return;}}}}被平台限流过的话无退避的死循环重连只会让封禁时间更长。四、解析层报文转事件协议差异全关在这一层演示用极简 JSON 协议真实平台换成对应解析实现即可接口不变classMessageParser{/** 返回 null 表示不是关心的事件心跳、进入房间等直接丢弃 */LiveEventparse(Stringjson){Stringtypefield(json,type);Stringuserfield(json,user);switch(type){casedanmaku:returnnewDanmakuEvent(user,field(json,text));casegift:returnnewGiftEvent(user,field(json,gift),num(json,count));caselike:returnnewLikeEvent(user,num(json,count));default:returnnull;}}// 演示用的字段提取生产环境请用 Jacksonnew ObjectMapper().readValue(json, Frame.class)privatestaticStringfield(Stringjson,Stringkey){Stringp\key\:\;intijson.indexOf(p);if(i0)return;intsip.length(),ejson.indexOf(,s);returne0?:json.substring(s,e);}privatestaticintnum(Stringjson,Stringkey){Stringp\key\:;intijson.indexOf(p);if(i0)return0;intsip.length(),es;while(ejson.length()Character.isDigit(json.charAt(e)))e;try{returnInteger.parseInt(json.substring(s,e));}catch(NumberFormatExceptionex){return0;}}}这一层的职责边界很清楚解析层是唯一知道平台协议长什么样的地方。接第二个平台就是多写一个 Parser其他层一行不改——这是事件驱动设计买来的扩展性。五、分发层有界队列 背压丢弃全文最关键的一段importjava.util.concurrent.*;importjava.util.function.Consumer;classEventBus{/** 有界队列容量 1 万。永远不用无界队列——OOM 都是无界队列堆出来的 */privatefinalBlockingQueueLiveEventqueuenewArrayBlockingQueue(10_000);privatefinalMapClass?,ListConsumerLiveEventlistenersnewConcurrentHashMap();voidon(Class?type,ConsumerLiveEventfn){listeners.computeIfAbsent(type,k-newCopyOnWriteArrayList()).add(fn);}/** 接收线程调用队列满时丢最旧的保证不阻塞 IO 线程 */voidpublish(LiveEvente){while(!queue.offer(e)){queue.poll();// 丢最旧。对直播数据旧的没价值}}voidstart(){ThreadtnewThread(()-{try{while(true){LiveEventequeue.take();// 阻塞取ListConsumerLiveEventfnslisteners.get(e.getClass());if(fnsnull)continue;for(ConsumerLiveEventfn:fns){try{fn.accept(e);}catch(Exceptionex){// 单个监听器异常不影响其他System.err.println(监听器异常: ex);}}}}catch(InterruptedExceptionignored){}},event-bus-dispatcher);t.setDaemon(true);t.start();}}这段的三个决策每个都值得单独说为什么单线程分发而不是线程池弹幕统计、礼物榜单这类场景要求同类型事件保序。单线程天然保序真要提吞吐按「房间 ID / 用户 ID」哈希到多个单线程队列别直接上无序的线程池。为什么丢弃最旧而不是丢最新榜单、热度这类统计要的是「最新状态」旧数据失效了。反过来如果你在做合规存档那改成 offer 失败就落盘再丢——策略由业务定有界 有策略这两个字不能让。监听器异常必须 try 住。一个统计模块抛 NPE不能把整条分发线程打死。六、组装与实测输出把三层拼起来完整代码可拼成单文件运行publicclassDanmakuDemo{publicstaticvoidmain(String[]args)throwsException{EventBusbusnewEventBus();MessageParserparsernewMessageParser();bus.start();bus.on(DanmakuEvent.class,e-{DanmakuEventd(DanmakuEvent)e;System.out.println([弹幕] d.user: d.text);});bus.on(GiftEvent.class,e-{GiftEventg(GiftEvent)e;System.out.println([礼物] g.user 送出 g.gift xg.count);});bus.on(LikeEvent.class,e-{LikeEventl(LikeEvent)e;System.out.println([点赞] l.user l.count);});// 模拟收到的平台原始报文真实场景来自 WebSocket onText 回调String[]frames{{\type\:\heartbeat\,\ts\:1},// 会被解析层丢弃{\type\:\danmaku\,\user\:\小张\,\text\:\主播讲的不错\},{\type\:\gift\,\user\:\老王\,\gift\:\火箭\,\count\:1},{\type\:\danmaku\,\user\:\阿七\,\text\:\已三连\},{\type\:\like\,\user\:\路人甲\,\count\:5}};for(Stringf:frames){LiveEventeparser.parse(f);if(e!null)bus.publish(e);}Thread.sleep(500);// 等分发线程消费完ReconnectPolicypnewReconnectPolicy();System.out.println(连续失败后的重连间隔: p.next(), p.next(), p.next(), p.next() 秒);}}本机JDK 21编译运行的真实输出[弹幕] 小张: 主播讲的不错 [礼物] 老王 送出 火箭 x1 [弹幕] 阿七: 已三连 [点赞] 路人甲 5 连续失败后的重连间隔: 1, 2, 4, 8 秒注意第一帧heartbeat没有出现在输出里——解析层正确地把它丢了。逐帧打印时多留心这种「该进的进、该挡的挡」比只看 happy path 更能确认链路是通的。编译运行方式JDK 11无任何第三方依赖javac-encodingUTF-8 DanmakuDemo.javajavaDanmakuDemo七、从 Demo 到生产还差这几步事项做法协议解析换 JacksonObjectMapper复用单例二进制协议先解压再拆包落库监听器里攒批每 200 条或每秒 flush 一次单条 insert 扛不住高峰分布式单机扛不住时采集进程只做「接入 转发」原始事件进 Kafka下游各自消费指标至少暴露三个数队列当前长度、丢弃总数、重连次数——丢数和断连肉眼是看不见的合规只接公开数据、遵守平台频率限制、用户昵称等展示类信息不落敏感库最后说回开头的那个判断这个框架真正值钱的不是「能连上弹幕」是队列满时的那一句queue.poll()。直播类系统所有线上事故一大半都出在「以为下游永远消费得过来」。