• 不用一直问，有消息直接推。怎么说呢，
• 延迟从秒级降到毫秒级。
• 服务器压力大幅降低。

WebSocket 原理

The protocol upgrades a single HTTP handshake into a persistent TCP connection。allowing bidirectional data flow without repeated handshakes.

HTTP 握手：
说到客户端，我要升级成 WebSocket
至于服务器，同意，协议切换
WebSocket 通信：
至于客户端，随时发消息 → 服务器
至于服务器，随时推消息 → 客户端

三、单机版弹幕程序

服务端主要代码

#1 导入依赖：

io.netty
netty-all
4.1.94.Final

#2 启动 WebSocket 服务：

@Component
@Slf4j
public class WebSocketServer {
@Value
private int port;@PostConstruct
public void start throws InterruptedException {
EventLoopGroup bossGroup = new NioEventLoopGroup;说起来，// 接受连接
EventLoopGroup workerGroup = new NioEventLoopGroup;// IO 处理
ServerBootstrap bootstrap = new ServerBootstrap;bootstrap.group
.channel
.childHandler);bootstrap.bind.sync;log.info，}
}

#3 初始化器：

public class WebSocketServerInitializer extends ChannelInitializer {
@Override
protected void initChannel {
ChannelPipeline pipeline = ch.pipeline;pipeline.addLast);// HTTP 编解码
pipeline.addLast);// HTTP 聚合器
pipeline.addLast);// WS 握手
pipeline.addLast);// 自定义业务处理器
}
}

#4 主要业务处理器：

@Component
@Slf4j
public class DanmakuHandler extends SimpleChannelInboundHandler {
// 房间 -> 所有使用者的 Channel 集合
private static final Map ROOM_CHANNELS = new ConcurrentHashMap<>;// Channel -> 使用者 ID / 房间 ID 映射，用于下线清理
private static final Map CHANNEL_USER = new ConcurrentHashMap<>;private static final Map CHANNEL_ROOM = new ConcurrentHashMap<>;@Autowired
private DanmakuService danmakuService;/** 收到前端发送的 JSON 消息 */
@Override
protected void channelRead0 {
String text = frame.text;Channel channel = ctx.channel;try {
DanmakuMessage msg = JSON.parseObject;switch ) {
case "join":
handleJoin;break，case "danmaku":
handleDanmaku;break，说起来，case "ping":
sendPong;break，default:
log.warn);}
} catch {
sendError;}
}
/** 使用者加入房间 */
private void handleJoin {
String roomId = msg.getRoomId;String userId = msg.getUserId;CHANNEL_USER.put;CHANNEL_ROOM.put;其实，ROOM_CHANNELS.computeIfAbsent(roomId。k -> ConcurrentHashMap.newKeySet).add;// 推送最近历史弹幕给新人
List history = danmakuService.getRecent;history.forEach);按理说，sendMessage(channel。new DanmakuMessage);log.info，}
/** 处理普通弹幕 */
private void handleDanmaku {
String roomId = msg.getRoomId;String userId = msg.getUserId;String content = msg.getContent;if ) {
sendError;return，0 }
if ) {
sendError;return，}
msg.setTimestamp);msg.setSender;danmakuService.saveAsync;// 异步落库
broadcastToRoom;// 广播给同房其他使用者
}
private void broadcastToRoom(String roomId。DanmakuMessage msg,Channel exclude) {
Set channels=ROOM_CHANNELS.get;if)return，String json=JSON.toJSONString;for{
if){
ch.writeAndFlush);}
}
}
/** 链路断开时资源清理 */
@Override
public void handlerRemoved {
Channel channel=ctx.channel;String room=CHANNEL_ROOM.remove;String uid=CHANNEL_USER.remove;if{
Set set=ROOM_CHANNELS.get;if{
set.remove;if){
ROOM_CHANNELS.remove;}
}
}
log.info;}
/** 辅助发送方法 */
private void sendMessage{
if){
channel.writeAndFlush(
new TextWebSocketFrame));}
}
private void sendError{
sendMessage);}
private void sendPong{
sendMessage);}
}

前端连接

// 建立连接
const ws=new WebSocket;// 连上后加入房间
ws.onopen=function{
ws.send(JSON.stringify({
type这方面。'join',roomId:'live_1001',userId:getUserId
}));}，怎么说呢，// 收到弹幕后渲染到屏幕上
ws.onmessage=function{
const data=JSON.parse;if{
showDanmaku;}
}，// 发弹幕后立即回执给使用者感官反馈
function sendDanmuku{
ws.send(JSON.stringify({
至于type。'danmaku',roomId:'live_1001',userId:getUserId,content:text
}));}
// 自动重连机制
ws.onclose=function{
setTimeout;// 3 秒后重新建立链接
};按理说，

单机能撑多少人？

Pain point: #使用者爆满时程序崩溃是最常见的恐慌点！保证单机承载能力是首要任务。

• No‑SQL 调优 + Netty 高性能 I/O → 单台  -10 万个并发 WebSocket 连接。
• If 超过此阈值，需要水平扩容 → 集群方案见下文。

四、集群部署——一台不够怎么办？

从问题来看，消息怎么跨机器广播？

#痛点 — "跨机器的使用者看不到彼此的弹幕",导致互动失效。

方案对比

方案一这方面，Redis Pub/Sub

@Component
public class RedisMessageRouter {
@Autowired
private StringRedisTemplate redisTemplate;/** 发布 */
public void publish {
String chan nel ="dan m a ku :room:"+room;redisTemplate.convertAndSend);}
/** 每个实例启动时订阅 */
@PostConstruct
public void subscribe{
// 使用 MessageListenerAdapter + RedisMessageListenerContainer
// …省略具体实现，}
}

• 快速上线，无需额外组件；适合 MVP 阶段，
• 若节点掉线，可配合 HA Redis 防止短暂丢失。不过，

方案二 ： 消息队列

@Component
public class Dan m a ku M Q P roducer {
@Autowired
private RocketMQTemplate rocketMQTemplate;public void sendDan m a ku{
rocketMQTemplate.convertAndSend;} }

@Component
public c lass D anmu ka M Q C onsumer {
@RocketMQMessag eListen er( topic ="dan mu ka _topic ",consumerGroup =" danmuka _group "。
messageModel = MessageModel.BROADCASTING )
public clas s Consumer implements RocketMQListener {
@Override
public voi d onMess age{
// 本机广播
dan m ak uHan dler.broadcastLocal,msg );}
} }
**Why choose this?**
* **Reliability** – 支持持久化、重试、事务。怎么说呢，* **Scalability** – 随机器水平扩容。仍保持全局一致性，

💡选择建议：


小项目 ：Redis Pub/Sub 足够。老实说，

大项目 ：Kafka / RocketMQ 更安全、更可



/ div>

五、弹幕怎么存？冷热分离 

数据特点 


越新的越关键；越旧的几乎没人看，

最近500 条需要即时读取；历史记录用于事后分析，老实说，



两级存储架构 

使用者发弹幕 ──► 热数据 Redis
│ │─► 保存最近500条，毫秒级读写
│ └─► 1 小时后自动过期
▼
冷数据 MySQL
│─► 异步写入。可做离线统计
/ code>

Redis 存储 

@Component
public class Dan m a kuSto rag e {
@Autowired
private Str ingRe disT temPlate redisTempl ate;private static final S tring KEY_PREFIX =" danmu ka :room:";private static final int MAX_RECENT =500;怎么说呢，public v oid saveRecent{
String key=KEY_PREFIX+roo mI d;String val=JSON.toJSONString;double score=ms g .getTi mestamp;redisTempla te.opsForZSet.add;redisTempla te.opsForZSet.removeRange;redisTempla te.expire;}
public L ista ll= redisTempLate.opsForZSet
.reverseRange;return al l .stream
.map)
.collect);}
}
/

MySQL 存储 
sql
CREATE TABLE danmu_ka_record (
id BIGINT PRIMARY KEY AUTO_INCREMENT。room_id VARCHAR NOT NULL COMMENT '房间ID',---省略字段---
) ENGINE=InnoDB;

异步持久化
Pain point: #同步写 MySQL 会瞬间把主线程堵死，引起卡顿或宕机！难道不该异步落库吗，
text
使用者 → Redis 快速写入 → 返回成功
│
└─► 推送至 MQ
Consumer 批量拉取 → 批量 INSERT MySQL
java
@Component
public class Dan mu kaPers istConsumer {
@Autowired
private Dan mu kaRecordMapper mapper;@RocketMQMessag eListener(topic ="da nmu_k_a_persist_topic",consumerGroup ="persist_grou p")
public class Consu mer implements RocketMQListener{
@Override
public vo i d onMess age{
Lists recs=
msgs.stream
.map
.collect);for，i+=1000){
int end=Math.min);mapper.batchInsert);}
}
}

六、防刷屏与敏感词过滤 

限流——滑动窗口实现 
java
@Component
public class RateLim itter {
@Autowired
private Str ingRed isTemplate redisTempl ate;public boolean allow{
long now=System.cur rentTimeMillis;long start=now-winSec*1000L;String key="ratelimit:"+uid;String lua=
"redis.call;"+
"local cnt=redis.call;"+
"if cnt < tonumber n "+   
"redis.call;"+   
"redis.call;"+   
"return 1 else return 0 end";
String member=now+\"-\"+UUID.randomUUID.toString.substring;
Long res=redisTemplat e.execute(
new DefaultRed isScript<>,Collections.singletonList,String.valueOf。String.valueOf,String.valueOf,member,String.valueOf);return res，=null && res==1L;}}
Pain point mitigation:


每秒最多5条，瞬时峰值也不会冲垮后台。老实说，

基于 Lua 脚本保证原子性，无竞争风险。



DFA 敏感词过滤 ​​​​​​​​​​​​​​​​​​​​​​​​​
java
@Component
public clas s SensitiveWordFilter{
private final Map dfaMap=new HashMap<>;@PostConstruct
public vo i d init{
Listwor ds=Arrays.asList("aodai"。"jinrengou",&quo t;"，buildDFA;}
privat e vo i d buildDFA{
for{
Map cur=dfaMap;for){
cur=c urrentMap.computeIfAbsent);}
cur.put)，}
}
pub lic boolea n contains{
if)retur n false;for，i++){
Map cur=dfaMap;for，j++){
char c=t xt.charAt;Map nxtur cur.ge t;if break，if) ret urn true;cur=nxt，}
} return false;}
}
Pain point: #未经过滤的敏感词会导致网站违规，被封禁！必须在业务层面做严密检查。

七、前端显示弹幕

两种渲染方式对比