news 2026/11/4 14:10:29

WebSocket+Kafka构建高并发实时消息通知系统

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
WebSocket+Kafka构建高并发实时消息通知系统

1. 项目背景与核心需求

现代互联网应用中，消息通知系统已经成为用户交互的核心组件之一。从社交软件的未读小红点到电商平台的订单状态变更，再到协同办公工具的实时协作提醒，一个高效可靠的消息通知系统直接影响着用户体验和产品粘性。

传统轮询（Polling）方式在实时性和服务器资源消耗上存在明显短板。我们需要的是一套能够实现：

  • 毫秒级实时推送
  • 百万级并发连接稳定维持
  • 离线消息可靠存储与补发
  • 消息状态精准追踪（已读/未读）
  • 水平扩展能力

这套系统我曾在多个千万级DAU项目中落地验证，今天就来拆解基于WebSocket+Kafka+Redis+TiDB的技术方案实现细节。

2. 技术架构全景图

2.1 组件分工与数据流

客户端 ─WebSocket─▶ 推送网关 ─┬─▶ Kafka（事件队列） ├─▶ Redis（在线状态/未读数） └─▶ TiDB（持久化存储）

各组件核心职责：

  • WebSocket网关：维护长连接，处理协议转换
  • Kafka：解耦业务系统与推送系统，作为消息缓冲区
  • Redis：存储用户在线状态、未读消息ID集合
  • TiDB：全量消息持久化，支持复杂查询

2.2 为什么选择这组技术栈？

WebSocket vs 长轮询

  • 单连接双向通信
  • 协议头开销仅2-10字节
  • 服务端可主动推送
  • 天然支持多路复用

Kafka作为消息中枢

  • 百万级TPS吞吐保障
  • 分区消费实现水平扩展
  • 消息保留策略可配置
  • Exactly-Once语义支持

Redis的独特价值

  • 5种数据结构应对不同场景
  • 单节点10万+ QPS
  • Lua原子脚本支持
  • 原生发布/订阅机制

TiDB的混合优势

  • MySQL协议兼容
  • 自动水平分片
  • 强一致性事务
  • HTAP混合负载能力

3. 核心模块实现细节

3.1 WebSocket网关设计

连接管理
// 使用gorilla/websocket库示例 var upgrader = websocket.Upgrader{ ReadBufferSize: 1024, WriteBufferSize: 1024, CheckOrigin: func(r *http.Request) bool { return true // 生产环境需严格校验 }, } func handleConnection(w http.ResponseWriter, r *http.Request) { conn, err := upgrader.Upgrade(w, r, nil) if err != nil { log.Println("Upgrade failed:", err) return } defer conn.Close() // 注册到连接管理器 client := NewClient(conn) connectionManager.Register(client) // 启动读写协程 go client.writePump() go client.readPump() }

关键参数调优：

  • 心跳间隔：建议25-30秒（避免Nginx默认60秒超时）
  • 读写缓冲区：根据消息体大小动态调整
  • 最大并发连接数：需测试OS文件描述符限制
消息协议设计
{ "msg_id": "uuidv4", "type": "NOTIFICATION|COMMAND|ACK", "payload": {}, "timestamp": 1630000000 }

重要提示：务必实现消息ID去重机制，防止网络重传导致重复处理

3.2 Kafka消息处理

分区策略选择

按用户ID哈希分区，保证同一用户的消息有序性：

// Producer配置 props.put("partitioner.class", "io.confluent.kafka.partitioner.UserIdPartitioner");

消费者组设计：

  • 推送组：实时消费消息进行推送
  • 备份组：异步持久化到TiDB
  • 监控组：生成统计指标

3.3 Redis存储模型

数据结构选型

  1. 在线状态：Hash

    • Key:user:status:{uid}
    • Field:conn_id,last_active

  2. 未读消息：Sorted Set

    • Key:user:unread:{uid}
    • Score: 消息时间戳
    • Value: 消息ID

  3. 全局消息索引：String

    • Key:msg:content:{msg_id}
    • Value: 消息内容JSON

Lua脚本示例（标记已读）：

local key = KEYS[1] local msg_id = ARGV[1] redis.call('ZREM', key, msg_id) return redis.call('ZCARD', key) -- 返回剩余未读数

3.4 TiDB表设计

CREATE TABLE notifications ( id BIGINT PRIMARY KEY AUTO_RANDOM, user_id BIGINT NOT NULL, msg_id VARCHAR(36) NOT NULL, content JSON NOT NULL, status TINYINT COMMENT '0-未发送 1-已发送 2-已读', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_user_status (user_id, status), UNIQUE KEY uk_msg (msg_id) ) PARTITION BY HASH(user_id) PARTITIONS 16;

读写分离配置：

tidb: read-only: true endpoints: - tidb-reader-1:4000 - tidb-reader-2:4000

4. 关键业务流程实现

4.1 实时推送流程

  1. 业务系统发送事件到Kafka
  2. 推送服务消费消息，查询Redis获取用户连接
  3. 通过WebSocket推送消息
  4. 更新Redis中的未读集合
  5. 异步写入TiDB

异常处理：

  • 用户离线时直接存入未读集合
  • 推送失败进入死信队列重试

4.2 未读消息补发

func fetchUnread(uid string) []Message { // 1. 获取未读消息ID集合 msgIDs := redis.ZRange("user:unread:"+uid, 0, -1) // 2. 批量获取消息内容 pipeline := redis.Pipeline() for _, id := range msgIDs { pipeline.Get("msg:content:" + id) } contents := pipeline.Exec() // 3. 转换为消息对象 var msgs []Message for _, c := range contents { msgs = append(msgs, parseMessage(c.(string))) } return msgs }

4.3 已读回写机制

客户端确认协议：

message ReadAck { string msg_id = 1; int64 user_id = 2; int64 timestamp = 3; }

服务端处理逻辑：

  1. 验证消息归属关系
  2. 原子化更新Redis未读集合
  3. 异步更新TiDB状态
  4. 可选发送回执给消息生产者

5. 性能优化实战

5.1 WebSocket集群部署

连接分布策略：

upstream ws_cluster { hash $http_user_id consistent; server ws1:8000; server ws2:8000; server ws3:8000; }

会话同步方案：

  • 定期将连接信息同步到Redis
  • 使用Redisson的RTopic实现跨节点通知

5.2 Kafka消费优化

// 批量消费配置 props.put("max.poll.records", 500); props.put("fetch.max.bytes", 52428800); // 消费者线程池 ExecutorService workers = Executors.newFixedThreadPool( Runtime.getRuntime().availableProcessors() * 2 );

5.3 Redis内存控制

  1. 设置未读集合上限：
    if redis.call('ZCARD', KEYS[1]) > 1000 then redis.call('ZREMRANGEBYRANK', KEYS[1], 0, -1001) end
  2. 启用内存淘汰策略：
    config set maxmemory-policy volatile-lru

6. 监控与运维

6.1 核心指标监控

指标名称采集方式报警阈值
活跃连接数Prometheus+Grafana> 80%容量
推送延迟P99Kafka Lag监控> 500ms
未读数积压Redis ZCARD单个用户>100
TiDB查询耗时TiDB DashboardSELECT>200ms

6.2 日志规范

2023-08-20T14:30:45Z INFO [PushWorker] uid=12345 msg_ids=[a1b2c3] status=delivered latency=128ms 2023-08-20T14:30:46Z WARN [KafkaConsumer] partition=3 lag=350 threshold=300

6.3 混沌工程测试

  1. 模拟网络分区：
    # 随机断开30%的Pod网络 chaosblade create network loss --percent 30 --interface eth0
  2. Kafka Broker宕机测试
  3. Redis主从切换演练

7. 典型问题排查实录

7.1 消息重复推送

现象：用户收到相同消息多次排查：

  1. 检查Kafka消费者提交的offset
  2. 验证消息ID生成规则
  3. 查看WebSocket ACK机制解决：实现服务端幂等处理
func isDuplicate(msgID string) bool { return redis.SetNX("dedup:"+msgID, "1", 24*time.Hour).Err() != nil }

7.2 连接频繁断开

现象：客户端每60秒重连根因：Nginx默认proxy_read_timeout为60秒优化：

location /ws { proxy_pass http://ws_backend; proxy_read_timeout 3600s; proxy_send_timeout 3600s; }

7.3 未读数不一致

场景：TiDB与Redis计数偏差解决方案：

  1. 实现双写校验机制
  2. 定期全量校对任务
  3. 提供强制刷新接口

这套系统在日活千万级的社交应用中稳定运行，峰值时承载200万+并发连接，日均处理20亿+消息。核心在于合理利用各组件优势：WebSocket解决实时性问题，Kafka保证可靠性，Redis保障性能，TiDB处理持久化与复杂查询。

版权声明: 本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系邮箱：809451989@qq.com进行投诉反馈，一经查实，立即删除！
网站建设 2026/11/4 14:09:12

OpenClaw大龙虾：模块化职场自动化工具实战解析

1. 项目背景与核心价值 OpenClaw大龙虾这个项目名称本身就充满了技术极客的幽默感——用海洋生物命名开源工具是开发者社区的传统艺能。这个2026年横空出世的项目之所以被称为"最炸裂开源项目"，是因为它完美解决了职场效率工具长期存在的矛盾：…

作者头像 李华
网站建设 2026/11/4 14:09:11

鼠标连点工具怎么做才顺手：时间精度、响应延迟与误触控制

聊鼠标连点工具这件事，我原本觉得没什么可写的。市面上一搜一大把，几十 KB 的小程序，双击打开、设个间隔、按个热键就完事，能有什么门槛？直到我自己动手做一个用得顺手的版本，才发现真正难的不是"让它…

作者头像 李华
网站建设 2026/11/4 14:08:23

TDX-PITE实验教程：8255/8254/8251硬件实操与FPGA引脚级排错指南

简介：本资源是面向高校《微机原理与接口技术》课程的配套实验教程，专为电子信息、自动化、计算机等工科专业本科生设计，聚焦80X86微处理器与外部硬件交互的工程实践能力培养。教程由西安唐都科教仪器公司官方编写，涵盖FPGA单元基础…

作者头像 李华
网站建设 2026/11/4 14:08:12

2026年全球企业服务商TOP7评测与选型指南

1. 项目背景与核心价值2026年的企业服务市场已经进入精细化运营阶段，各类第三方服务商如雨后春笋般涌现。对于GEO（全球企业主）而言，如何在数百家服务提供商中快速锁定最适合自身业务需求的合作伙伴，成为影响运营效率的…

作者头像 李华
网站建设 2026/11/4 14:07:18

Windows版Microsoft Teams手工安装实战指南

1. 为什么还要“手工安装”Teams？这事儿真没你想的那么简单最近在几个技术交流群里，频繁看到有人问：“Teams不是微软自家软件吗？官网下载exe点两下不就装好了？还搞什么手工安装？”——这话乍一听挺有道理&a…

作者头像 李华
网站建设 2026/11/4 14:04:37

数据降维技术解析：从原理到实践应用

1. 数据降维的核心价值与挑战 在处理现代大数据项目时，我们经常会遇到"维度灾难"这个拦路虎。想象一下你正在分析一个包含1000个特征的用户行为数据集——不仅计算资源消耗呈指数级增长，更重要的是许多特征之间存在冗余和噪声，反而…

作者头像 李华