跟着大厂学架构01：如何利用开源方案，复刻B站那套“永不崩溃”的评论系统？
引子
本文将基于哔哩哔哩技术团队分享的《B 站评论系统的多级存储架构》，这篇文章为我们揭示了，在亿级流量冲击下，一个顶流社区的核心互动功能是如何通过精妙设计保持高可用的。
考虑到其架构中的核心组件“泰山 KV 存储”为 B 站自研，我们无法直接获取。因此，本文的核心目标，便是将这套先进的私有架构“翻译”成一套人人皆可上手的开源实现。我们将以原文为蓝本，聚焦于如何利用我们更熟悉的 MySQL、Redis 及 Java 技术栈，完整复刻其多级存储、数据同步与对冲降级等关键设计。
希望这篇文章能成为你探索大厂架构的起点。文中若有任何见解或疑问，期待在评论区与你一同探讨，共同进步。
B站评论系统架构解析
（ B站的主存储使用的是 TiDB，本文为了保持”原汁原味“，也采用相同的数据库。各位读者如果想复刻，可以使用自己熟悉的数据库，比如 MySQL 。因为架构理论是通用的，此部分核心思想不变。）
为了方便理解，我们把B站庞大的评论系统想象成一个超大型的图书馆，里面数不清的评论就是一本本书。
核心痛点
找书太慢 (查询性能瓶颈)：当成千上万的读者都想看“最受欢迎的10本书”（也就是点赞最多的10条评论）时，图书馆馆长（TiDB数据库）就得把所有书架翻个底朝天，再一本本排序，找出TOP 10。这个过程不仅慢得让人抓狂，还会把馆长累垮，导致其他借书、还书的请求（常规读写）也跟着变慢。
馆长病了怎么办 (单点故障风险)：图书馆就这么一个馆长（TiDB集群）。万一他生病了、罢工了（比如部分节点宕机或网络抖动），整个图书馆岂不是要关门大吉？
为了解决这两个问题，B站的技术同学们设计了一套巧妙的“多级存储”系统，核心思想是：为不同的任务，设置不同的“查询路径”。不让主数据库（TiDB）承担所有工作，而是引入了一个速度极快的副手（在我们的方案里是Redis，原文用的是自研的泰山KV）。
关键设计1：分开“书”和“目录”，实现闪电查询
这是整个架构的灵魂。面对“找书太慢”的问题，他们不再让读者通过馆长直接在书架上找书，而是设立了一个专门的“目录查询台”。这里采用的是Index + KV 的存储模型：
KV（Key-Value）：每一本书（即每条评论）的内容本身，都存放在一个巨大的仓库里。每本书都有一个独一无二的编号 Key（reply_id），只要报出编号，就能瞬间拿到书（Value）。
Index（索引）：为了能快速找到“最受欢迎的书”或“最新的书”，图书馆专门聘请了一位“目录管理员”（在我们的方案里就是 Redis）。他手里并不存书，只保管着几份实时更新的“目录索引卡”（也就是 Redis 的 Sorted Set），比如：一张是“热度榜”，按点赞数从高到低排列着所有书的编号；一张是“新书榜”，按出版时间从新到旧排列着所有书的编号。
现在，当一个读者想看“点赞最高的10本书”时，流程就变得极其高效了：
读者前往“目录查询台”，向“目录管理员”索要点赞最高的10本书的目录。
目录管理员（Redis）将“热度榜”索引卡上排名前10的书的编号告诉了读者。
读者拿着这10个编号，去KV仓库，精准地取出了这10本书。
整个过程完全避免了去遍历所有书架的笨重操作。这就是 Index + KV 模型 的精髓——通过索引快速定位，再通过Key快速获取，让查询性能发生了质的飞跃。
关键设计二：用“工作日志”和“公告栏”，让目录保持最新
好了，我们现在有了一个在“目录查询台”工作、反应神速的“目录管理员”（Redis）。但他远在查询台，怎么能实时知道馆长（TiDB）在后台书库里做了什么呢？比如：
书库里新上架了一批书？
某本书的口碑爆了，被读者疯狂点赞？
总不能让目录管理员每隔几分钟就跑去后台问一遍吧？那样效率太低了。
图书馆的解决方案是建立一套自动化的内部信息同步系统。这套系统依赖两样东西：总馆长的“工作日志”和部门间的“电子公告栏”。
工作流程是这样的：
总馆长记录日志（写 Binlog） 总馆长（TiDB）有一个雷打不动的习惯：他在后台书库做的任何操作（新书入库、为某本书增加点赞数），都会立刻、按顺序、一字不差地记录在一本公开的《每日工作日志》上。这本日志，就是我们的 binlog，它是图书馆所有操作的绝对事实源头。
助理复印并张贴公告（TiCDC 捕获变更） 总馆长有一位专门的助理，名叫 TiCDC。这位助理不干别的，他的唯一任务就是趴在日志旁边，当一个“复印机”。只要总馆长写下新的一行，TiCDC 助理就立刻复印一份，做成一张“工作简报”。
信息上公告栏（发送到 Kafka） TiCDC 助理会把这张“工作简报”立刻贴到图书馆内部的“电子公告栏”上。这个公告栏就是我们的消息队列 Kafka，所有部门的职员都能看到上面的信息滚动。
目录管理员更新卡片（消费消息） 现在，在前台的目录管理员（Redis）就轻松了。“电子公告栏”上面出现了新的简报（比如：“《xxx评论》点赞数+1”）并提醒后，他就知道该如何更新自己手里的“热度榜”索引卡了。
这个过程，技术上称之为“异步”。意思是，从总馆长记录日志，到目录管理员看到公告，中间会有一个微乎其微的时间差（通常是毫秒级）。但这样做的好处是，总馆长可以专心处理核心的图书管理工作，完全不用停下来等目录管理员确认“收到”，整个图书馆的运行效率因此大大提升。
关键设计三：三道“保险”，防止公告系统出差错
我们建立的这套“工作日志 + 公告栏”的同步系统虽然高效，但毕竟是自动化的，万一出点意外怎么办？比如：
公告栏系统（Kafka）偶尔抽风，给目录管理员的“提醒”发送失败了？
目录管理员（Redis）正好特别忙，没来得及处理收到的提醒，导致数据错过了？
如果这些小概率事件发生，目录管理员的索引卡就会和总馆长的日志对不上，数据就乱套了。为了保证数据最终绝对一致，图书馆设置了三道严密的“保险措施”：
险一：重要通知，使命必达（失败重试） “电子公告栏”（Kafka）系统非常智能。如果它发出一条提醒，但目录管理员没有确认“收到”（ACK），系统不会当它发送成功了。它会把这条通知标记为“待处理”，过一会儿重新发送，直到目录管理员确认收到并处理完毕为止。这就确保了任何一条重要的工作简报都不会丢失。
保险二：盖上“版本戳”，防止信息错乱（版本号 + CAS） 这是一个更微妙的问题：万一网络延迟，导致“点赞数到11”的提醒比“点赞数到10”的提醒先到，怎么办？目录管理员岂不是会先更新到11，又错误地用旧数据覆盖成10？
为了解决这个问题，总馆长（TiDB）在记录《工作日志》时，每次修改都会在旁边盖上一个递增的“版本戳”（version）。助理 TiCDC 在复印时会把这个版本戳也抄到公告上。
现在，目录管理员在更新索引卡时，必须遵守一条铁律：先检查提醒上的版本戳，是不是比自己卡片上记录的当前版本戳要新。只有新的才更新，旧的直接忽略。 这样就完美避免了数据被错误地“降级”回去。
保险三：每日盘点，确保万无一失（对账系统） 作为最后一道防线，图书馆有每日盘点的制度。每天闭馆后，总馆长（TiDB）和目录管理员（Redis）会坐在一起，拿出总馆长的《工作日志》和目录管理员的所有索引卡，进行一次关键数据的抽查核对。一旦发现有对不上的地方，就以总馆长的日志为准，立刻修正。这个过程就是“对账”，是数据最终一致性的终极保障。
有了这三道保险，整个数据同步系统就变得固若金汤，非常可靠了。
关键设计四：前台“兵分两路”，保证服务不中断
现在，图书馆几乎完美了。但我们还要考虑一个极端情况：如果总馆长（TiDB）今天闹肚子，处理后台请求特别慢，甚至暂时联系不上了，怎么办？难道让前台的读者干等着吗？
当然不。B站的办法非常聪明，他们没有把这个难题交给读者，而是为图书馆的“前台接待员”（代表我们的应用服务层）制定了一套巧妙的应急预案——对冲策略。
现在，当一个读者来到前台想借书时，接待员会这样做：
接待员接到请求后，首先通过内部系统联系后台的总馆长（TiDB），让他去书库找书。
但接待员并不会傻等。他一边等，一边在心里启动一个20毫秒的倒计时。
如果倒计时结束，总馆长那边还没消息，接待员会立刻兵分两路：他会同时向速度飞快的目录管理员（Redis）也发出请求，让他帮忙查询书的编号。
最后，总馆长和目录管理员，谁先返回结果，接待员就用谁的结果，立刻把书（或信息）交给读者。
这样一来，无论总馆长是健康、迟钝还是暂时“失联”，读者在前台的感受永远是“丝滑”的。因为接待员总能通过这条“对冲”的备用路线，从更快的渠道（绝大部分时候是目录管理员Redis）拿到结果。
这就是“对冲降级”，一种极其优雅、对用户体验无感知的容灾方案。
开源方案技术选型与架构搭建
通过对上述关键设计点进行分析，我们提出一套基于主流开源组件的技术实现方案。该方案旨在复刻B站评论系统的核心架构思想，实现高并发读写、低延迟查询与高可用性。
主存储: TiDB 集群，为最大程度还原原架构，选用其原生分布式数据库TiDB。其水平扩展能力能无缝支撑高并发场景。如果读者对MySQL更熟悉，可以将主存储替换为MySQL集群,数据同步工具可选用Debezium来替代TiCDC。
多级存储 : Redis Cluster，作为B站自研存储“泰山KV”的开源替代品。其高性能的Key-Value读写及Sorted Set数据结构，是实现缓存和热点排行的理想选择。
数据同步管道 (DTS替代品):
变更捕获: TiCDC，TiDB官方的增量数据同步工具，负责捕获binlog变更并将其推送到下游。
消息队列: Kafka，作为行业标准的消息中间件，用于解耦CDC组件与消费端应用，提供数据缓冲与高可用性保障。
同步消费程序: Spring Boot 应用，可使用spring-kafka库以消费消息，并通过 Lettuce 或 Jedis 等客户端将数据变更应用至Redis Cluster。
数据对账服务:分布式定时任务框架 (如 XXL-Job)，负责定期触发数据比对与修复流程，保障主存储与二级存储的最终一致性。
技术实现
理论最终要通过实践来检验。现在，我们将亲手搭建并实现这套评论系统的核心架构。
环境搭建
我们可以通过 Docker 快速地部署一个包含 TiDB、TiCDC、Kafka 和 Redis 的完整开发环境。使用 docker-compose 来编排这些服务，实现“一键启动”。首先可以创建一个项目目录，并在目录下新建一个docker-compose.yml 文件。
version: '3.8'
services:
# 1. TiDB 集群 (包含 PD, TiKV, TiDB)
pd:
image: pingcap/pd:v6.5.0
ports:
- "2379:2379"
volumes:
- ./pd_data:/data/pd
command:
- --name=pd
- --client-urls=http://0.0.0.0:2379
- --peer-urls=http://0.0.0.0:2380
- --advertise-client-urls=http://pd:2379
- --advertise-peer-urls=http://pd:2380
- --initial-cluster=pd=http://pd:2380
networks:
- tidb_net
tikv:
image: pingcap/tikv:v6.5.0
ports:
- "20160:20160"
volumes:
- ./tikv_data:/data/tikv
command:
- --addr=0.0.0.0:20160
- --advertise-addr=tikv:20160
- --pd=pd:2379
depends_on:
- pd
networks:
- tidb_net
tidb:
image: pingcap/tidb:v6.5.0
ports:
- "4000:4000" # MySQL 客户端连接端口
- "10080:10080"
command:
- --store=tikv
- --path=pd:2379
depends_on:
- tikv
networks:
- tidb_net
# 2. TiCDC (数据同步工具)
ticdc:
image: pingcap/ticdc:v6.5.0
ports:
- "8300:8300"
command: ["/cdc", "server", "--pd=http://pd:2379", "--addr=0.0.0.0:8300", "--advertise-addr=ticdc:8300"]
volumes:
- ./changefeed.toml:/tmp/changefeed.toml
depends_on:
- tidb
restart: unless-stopped
networks:
- tidb_net
# 3. Kafka 集群 (包含 Zookeeper 和 Kafka)
zookeeper:
image: confluentinc/cp-zookeeper:7.3.0
ports:
- "2181:2181"
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ZOOKEEPER_TICK_TIME: 2000
networks:
- tidb_net
kafka:
image: confluentinc/cp-kafka:7.3.0
ports:
- "29092:29092" # 暴露给宿主机(Spring Boot应用)的端口
depends_on:
- zookeeper
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
networks:
- tidb_net
# 4. Redis (二级存储)
redis:
image: redis:6.2-alpine
ports:
- "6379:6379"
networks:
- tidb_net
networks:
tidb_net:
driver: bridge
在项目根目录下创建 changefeed.toml 文件，作用是让TiCDC只需要盯着 comment_db 数据库下的 t_comment 表。
[filter]
# 过滤规则
rules = ['comment_db.t_comment']
接着在 docker-compose.yml 文件所在的目录下，执行以下命令：
docker-compose up -d
稍微等待一下，让所有服务完成启动。
我们还需要为TiCDC 创建数据同步任务，它的作用类似于 TiDB 和 Kafka 之间的数据纽带。
# 1. 进入 TiCDC 容器
docker-compose exec ticdc sh
# 2. 在容器内执行创建命令
/cdc cli changefeed create \
--pd="http://pd:2379" \
--changefeed-id="comment-sync-to-kafka" \
--sink-uri="kafka://kafka:9092/comment-topic?protocol=open-protocol&kafka-version=2.8.1&max-message-bytes=10485760&partition-num=1" \
--config="/tmp/changefeed.toml"
# 3. 退出容器
exit
至此，我们的开发环境已经准备就绪！
库表设计
我们可以使用任何标准的 MySQL 客户端（如 Navicat, DBeaver, 或者命令行）连接到 TiDB。连接信息如下：
Host: 127.0.0.1
Port: 4000
User: root
Password: (默认为空)
我们需要在 TiDB 中创建一张用于存储评论的表。执行以下 SQL 语句：
-- 创建数据库
CREATE DATABASE `comment_db` DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci;
-- 切换到该数据库
USE `comment_db`;
-- 创建评论表
CREATE TABLE `t_comment` (
`id` BIGINT(20) NOT NULL AUTO_RANDOM, -- TiDB 推荐使用 AUTO_RANDOM 防止写热点
`content` TEXT NOT NULL COMMENT '评论内容',
`user_id` VARCHAR(64) NOT NULL COMMENT '用户ID',
`likes` INT(11) UNSIGNED NOT NULL DEFAULT '0' COMMENT '点赞数',
`created_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
`updated_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
`version` INT(11) UNSIGNED NOT NULL DEFAULT '0' COMMENT '乐观锁版本号',
PRIMARY KEY (`id`),
KEY `idx_likes` (`likes`),
KEY `idx_created_at` (`created_at`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;
补充说明:
AUTO_RANDOM: 这是 TiDB 的一个特性，用于在分片时打散主键，避免写入集中在单个 TiKV 节点上，从而解决写热点问题。
version: 这个字段用于实现乐观锁，对应我们“关键设计三”中的版本号机制，防止并发更新时数据被覆盖。
应用开发
我们首先创建一个 Spring Boot 项目，并引入相关 Maven依赖。
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
接下来，在 application.yml 中配置好所有服务的连接信息。
server:
port: 8082
spring:
# --- TiDB ---
datasource:
url: jdbc:mysql://127.0.0.1:4000/comment_db?useUnicode=true&characterEncoding=utf-8&serverTimezone=Asia/Shanghai
username: root
password:
driver-class-name: com.mysql.cj.jdbc.Driver
jpa:
hibernate:
ddl-auto: update
show-sql: true
properties:
hibernate:
dialect: org.hibernate.dialect.MySQL8Dialect
# --- Redis ---
data:
redis:
host: 127.0.0.1
port: 6379
database: 0
# --- Kafka ---
kafka:
bootstrap-servers: localhost:29092
consumer:
group-id: comment-sync-group
auto-offset-reset: earliest
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer
### 调度中心部署根地址 [选填]：如调度中心集群部署存在多个地址则用逗号分隔。执行器将会使用该地址进行"执行器心跳注册"和"任务结果回调"；为空则关闭自动注册；
xxl:
job:
admin:
addresses: http://127.0.0.1:8080/xxl-job-admin
### 执行器通讯TOKEN [选填]：非空时启用；
accessToken: default_token
### 执行器AppName [选填]：执行器心跳注册分组依据；为空则关闭自动注册
executor:
appname: xxl-job-executor-sample
### 执行器注册 [选填]：优先使用该配置作为注册地址，为空时使用内嵌服务 ”IP:PORT“ 作为注册地址。从而更灵活的支持容器类型执行器动态IP和动态映射端口问题。
address:
### 执行器IP [选填]：默认为空表示自动获取IP，多网卡时可手动设置指定IP，该IP不会绑定Host仅作为通讯实用；地址信息用于 "执行器注册" 和 "调度中心请求并触发任务"；
ip: 127.0.0.1
### 执行器端口号 [选填]：小于等于0则自动获取；默认端口为9999，单机部署多个执行器时，注意要配置不同执行器端口；
port: 9999
### 执行器运行日志文件存储磁盘路径 [选填] ：需要对该路径拥有读写权限；为空则使用默认路径；
logpath: /data/applogs/xxl-job/jobhandler
### 执行器日志文件保存天数 [选填] ： 过期日志自动清理, 限制值大于等于3时生效; 否则, 如-1, 关闭自动清理功能；
logretentiondays: 30
Comment 实体定义评论的数据结构。我们使用了 @Version 注解来实现乐观锁，这是处理高并发点赞场景、防止数据不一致的关键。
package com.cc.entity;
import jakarta.persistence.*;
import lombok.Data;
import java.io.Serializable;
import java.time.LocalDateTime;
@Data
@Entity
@Table(name = "t_comment")
public class Comment implements Serializable {
private static final long serialVersionUID = 1L;
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;
private String content;
private String userId;
private Integer likes = 0;
@Column(name = "created_at", updatable = false)
private LocalDateTime createdAt;
@Column(name = "updated_at")
private LocalDateTime updatedAt;
@Version // 关键！标记为乐观锁版本号字段
private Integer version;
@PrePersist
protected void onCreate() {
this.createdAt = LocalDateTime.now();
this.updatedAt = LocalDateTime.now();
}
@PreUpdate
protected void onUpdate() {
this.updatedAt = LocalDateTime.now();
}
}
接着创建controller，作为API入口。
package com.cc.controller;
import com.cc.dto.CreateCommentRequest;
import com.cc.entity.Comment;
import com.cc.service.CommentService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import java.util.List;
@RestController
@RequestMapping("/api/comments")
public class CommentController {
@Autowired
private CommentService commentService;
// 发布评论
@PostMapping
public ResponseEntity
Comment comment = commentService.createComment(request.getContent(), request.getUserId());
return new ResponseEntity<>(comment, HttpStatus.CREATED);
}
// 点赞评论
@PostMapping("/{id}/like")
public ResponseEntity
try {
Comment comment = commentService.likeComment(id);
return ResponseEntity.ok(comment);
} catch (Exception e) {
// 这里可以更精细地处理乐观锁冲突异常
return ResponseEntity.status(HttpStatus.CONFLICT).build();
}
}
// 获取点赞最多的10条评论
@GetMapping("/hot")
public ResponseEntity
List
return ResponseEntity.ok(comments);
}
}
CommentService 承载了核心的业务逻辑，我们采用经典的缓存旁路模式来读取热点数据，并设计了优雅的降级策略。
package com.cc.service;
import com.cc.entity.Comment;
import com.cc.repository.CommentRepository;
import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.persistence.EntityNotFoundException;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
@Slf4j
@Service
public class CommentService {
private static final String KEY_HOT_COMMENTS = "hot_comments";
private static final String KEY_COMMENT_CACHE_PREFIX = "comment:";
@Autowired
private CommentRepository commentRepository;
@Autowired
private RedisTemplate
@Autowired
private ObjectMapper objectMapper;
@Transactional
public Comment createComment(String content, String userId) {
Comment comment = new Comment();
comment.setContent(content);
comment.setUserId(userId);
// 其他字段如 likes, version, createdAt, updatedAt 都有默认值或由 @PrePersist 处理
return commentRepository.save(comment);
}
@Transactional
public Comment likeComment(Long commentId) {
// 使用乐观锁来处理点赞
Comment comment = commentRepository.findById(commentId)
.orElseThrow(() -> new EntityNotFoundException("评论不存在: " + commentId));
comment.setLikes(comment.getLikes() + 1);
// save 方法在有 @Version 的情况下会自动检查版本号，如果冲突会抛出异常
return commentRepository.save(comment);
}
public List
try {
// 1. 优先从 Redis 的 Sorted Set 中获取 Top 10 的评论 ID
Set
if (commentIds == null || commentIds.isEmpty()) {
log.warn("Redis 热度榜为空，执行降级策略，直接查询数据库！");
return fetchFromDB();
}
// 2. 根据 ID 批量从 Redis 缓存中获取评论详情
List
.map(id -> KEY_COMMENT_CACHE_PREFIX + id)
.collect(Collectors.toList());
List
// 3. 反序列化并处理缓存未命中的情况
return cachedCommentsJson.stream()
.map(json -> {
try {
// 如果缓存命中，直接反序列化
if (json != null) {
return objectMapper.readValue((String) json, Comment.class);
}
// 如果缓存未命中（例如已过期），则需要回源到数据库查询
// 注意：此处的 ID 需要从 commentIds 中反查，为简化，此处返回 null
// 在生产环境中，需要设计更完善的回源机制
return null;
} catch (Exception e) {
log.error("从 Redis 反序列化评论失败", e);
return null;
}
})
.filter(java.util.Objects::nonNull)
.collect(Collectors.toList());
} catch (Exception e) {
log.error("从 Redis 获取热度榜失败，执行降级策略!", e);
// 降级：如果 Redis 发生任何异常，直接查询数据库
return fetchFromDB();
}
}
// 降级方法：直接从数据库查询
private List
return commentRepository.findTop10ByOrderByLikesDesc();
}
}
还要在repository中创建一些必备的方法。
package com.cc.repository;
import com.cc.entity.Comment;
import io.lettuce.core.dynamic.annotation.Param;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.data.jpa.repository.Query;
import org.springframework.stereotype.Repository;
import java.time.LocalDateTime;
import java.util.List;
@Repository
public interface CommentRepository extends JpaRepository
// 用于降级策略，从数据库直接获取热榜
List
// 用于定时校准任务，查询指定时间后的热点数据
@Query("SELECT c FROM Comment c WHERE c.updatedAt >= :since ORDER BY c.likes DESC")
List
}
CommentSyncConsumer 是连接 TiCDC 和 Redis 的关键枢纽。它监听 Kafka 主题，解析 TiCDC 发送的二进制消息，并将最新的数据变更实时同步到 Redis 的热榜和详情缓存中。
package com.cc.consumer;
import com.cc.entity.Comment;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.TimeUnit;
@Service
public class CommentSyncConsumer {
private static final Logger log = LoggerFactory.getLogger(CommentSyncConsumer.class);
private static final String KEY_HOT_COMMENTS = "hot_comments";
private static final String KEY_COMMENT_CACHE_PREFIX = "comment:";
@Autowired
private ObjectMapper objectMapper;
@Autowired
private RedisTemplate
// 定义TiDB时间格式的解析器，这个格式是 "yyyy-MM-dd HH:mm:ss"
private static final DateTimeFormatter TIDB_DATETIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
@KafkaListener(topics = "comment-topic", groupId = "comment-sync-group-final")
public void listen(@Payload(required = false) byte[] messageBytes) {
if (messageBytes == null || messageBytes.length == 0) {
// 心跳消息，正常忽略
return;
}
String originalMessage = new String(messageBytes, StandardCharsets.UTF_8);
String jsonPayload = "";
try {
int jsonStart = originalMessage.indexOf('{');
if (jsonStart == -1) {
log.warn("收到的消息不包含有效的JSON对象，已忽略。消息: {}", originalMessage);
return;
}
jsonPayload = originalMessage.substring(jsonStart);
log.info("清理后的有效JSON消息: {}", jsonPayload);
JsonNode rootNode = objectMapper.readTree(jsonPayload);
String eventType = null;
JsonNode dataNode = null;
if (rootNode.has("u")) {
eventType = "UPSERT";
dataNode = rootNode.get("u");
} else if (rootNode.has("d")) {
eventType = "DELETE";
dataNode = rootNode.get("d");
} else {
log.warn("收到的JSON格式未知 (既没有'u'也没有'd')，已忽略。消息: {}", jsonPayload);
return;
}
Comment comment = new Comment();
// 从每个包装对象中提取 'v' (value)
comment.setId(dataNode.get("id").get("v").asLong());
comment.setContent(dataNode.get("content").get("v").asText());
// --- 关键修正：适配实体类中的 userId ---
// JSON 中是 "user_id"，实体类中是 "userId"，需要手动对应
comment.setUserId(dataNode.get("user_id").get("v").asText());
comment.setLikes(dataNode.get("likes").get("v").asInt());
comment.setVersion(dataNode.get("version").get("v").asInt());
// --- 关键修正：将字符串转换为实体类所需的 LocalDateTime ---
String createdAtStr = dataNode.get("created_at").get("v").asText();
String updatedAtStr = dataNode.get("updated_at").get("v").asText();
comment.setCreatedAt(LocalDateTime.parse(createdAtStr, TIDB_DATETIME_FORMATTER));
comment.setUpdatedAt(LocalDateTime.parse(updatedAtStr, TIDB_DATETIME_FORMATTER));
log.info("成功解析出 Comment 对象: {}", comment);
// 后续业务逻辑完全不变
if ("UPSERT".equals(eventType)) {
if (comment.getId() != null) {
redisTemplate.opsForZSet().add(KEY_HOT_COMMENTS, comment.getId().toString(), comment.getLikes());
log.info("更新 Redis 热度榜: Comment ID = {}, Likes = {}", comment.getId(), comment.getLikes());
String cacheKey = KEY_COMMENT_CACHE_PREFIX + comment.getId();
String commentJson = objectMapper.writeValueAsString(comment);
redisTemplate.opsForValue().set(cacheKey, commentJson, 1, TimeUnit.HOURS);
log.info("更新 Redis 缓存: Key = {}", cacheKey);
}
} else if ("DELETE".equals(eventType)) {
if (comment.getId() != null) {
redisTemplate.opsForZSet().remove(KEY_HOT_COMMENTS, comment.getId().toString());
log.info("从 Redis 热度榜中移除: Comment ID = {}", comment.getId());
redisTemplate.delete(KEY_COMMENT_CACHE_PREFIX + comment.getId());
log.info("从 Redis 缓存中删除: Key = {}", KEY_COMMENT_CACHE_PREFIX + comment.getId());
}
}
} catch (Exception e) {
log.error("处理 Kafka 消息时发生未知异常! 原始消息: [{}], 尝试处理的JSON部分: [{}]", originalMessage, jsonPayload, e);
}
}
}
为了保证数据最终一致性，我们引入了 XXL-Job来每日执行一次数据校准任务，确保 Redis 中的数据与 TiDB 中的“黄金数据”保持最终一致。
package com.cc.jobhandler;
import com.cc.service.CommentSyncService;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class CommentSyncXxlJobHandler {
private static final Logger log = LoggerFactory.getLogger(CommentSyncXxlJobHandler.class);
@Autowired
private CommentSyncService commentSyncService;
@XxlJob("syncHotCommentsJobHandler")
public void executeSyncHotComments() {
XxlJobHelper.log("【热点评论同步任务】开始执行...");
log.info("【热点评论同步任务】XXL-Job 触发，开始执行...");
try {
// 调用核心业务逻辑
commentSyncService.syncHotCommentsToRedis();
XxlJobHelper.log("【热点评论同步任务】执行成功！");
log.info("【热点评论同步任务】执行成功！");
// 主动上报执行成功
XxlJobHelper.handleSuccess();
} catch (Exception e) {
XxlJobHelper.log("【热点评论同步任务】执行失败！错误信息: {}", e.getMessage());
log.error("【热点评论同步任务】执行失败！", e);
// 主动上报执行失败
XxlJobHelper.handleFail(e.getMessage());
}
}
}
我们遵循逻辑与调度分离的最佳实践：Handler 负责被 XXL-Job 触发，Service 负责执行具体的同步逻辑。
package com.cc.service;
import com.cc.entity.Comment;
import com.cc.repository.CommentRepository;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.TimeUnit;
@Service
public class CommentSyncService {
private static final Logger log = LoggerFactory.getLogger(CommentSyncService.class);
private static final String KEY_HOT_COMMENTS = "hot_comments";
private static final String KEY_HOT_COMMENTS_TEMP = "hot_comments_temp";
private static final String KEY_COMMENT_CACHE_PREFIX = "comment:";
@Autowired
private CommentRepository commentRepository;
@Autowired
private RedisTemplate
@Autowired
private ObjectMapper objectMapper;
@Transactional(readOnly = true)
public void syncHotCommentsToRedis() throws Exception {
log.info("开始执行热点评论数据同步业务逻辑...");
// 1. 定义同步范围
LocalDateTime sevenDaysAgo = LocalDateTime.now().minusDays(7);
PageRequest pageRequest = PageRequest.of(0, 1000); // Top 1000
// 2. 从TiDB查询
List
if (hotComments.isEmpty()) {
log.info("未发现需要同步的热点评论，任务结束。");
return;
}
log.info("从数据库查询到 {} 条热点评论进行同步。", hotComments.size());
// 3. 写入临时Key
redisTemplate.delete(KEY_HOT_COMMENTS_TEMP);
for (Comment comment : hotComments) {
redisTemplate.opsForZSet().add(KEY_HOT_COMMENTS_TEMP, comment.getId().toString(), comment.getLikes());
String cacheKey = KEY_COMMENT_CACHE_PREFIX + comment.getId();
String commentJson = objectMapper.writeValueAsString(comment);
redisTemplate.opsForValue().set(cacheKey, commentJson, 1, TimeUnit.HOURS);
}
// 4. 原子化重命名
redisTemplate.rename(KEY_HOT_COMMENTS_TEMP, KEY_HOT_COMMENTS);
log.info("成功同步 {} 条热点评论到 Redis！", hotComments.size());
}
}
至此，我们的仿哔哩哔哩评论系统就开发完成了！
小结
这套架构的核心在于 TiDB 作为数据存储基石与 Redis 作为高速缓存的协同工作。我们并未止步于简单的缓存读写，而是设计并实现了一套 “实时同步 + 定时校准” 的双轨数据同步机制：
实时同步：通过 TiDB -> TiCDC -> Kafka -> 服务消费 的链路，实现了数据变更的毫秒级捕获与推送，确保了用户交互的即时性。
定时校准：借助 XXL-Job 的分布式调度任务，对数据进行周期性校对，以此作为最终一致性的坚实保障，有效应对了消息丢失或服务中断等异常情况。
在实现层面，我们采用了 Spring Boot 技术栈，并结合了乐观锁、缓存降级、原子化更新等设计模式，确保了业务逻辑的健壮性。
文章中的代码已上传到Github，感兴趣的读者可以自行查看：https://github.com/Pitayafruits/copy-bilibili-reviews-system