用 Spring Boot + Redis 实现哔哩哔哩弹幕系统（上篇博客改进版）

用 Spring Boot + Redis 实现哔哩哔哩弹幕系统

支持：历史弹幕 + 实时弹幕 + 敏感词过滤 + 限频 + 持久化

🧩 项目功能总览

🧱 技术栈

🗃️ 弹幕数据模型（MySQL）

CREATE TABLE danmu (
  id BIGINT AUTO_INCREMENT PRIMARY KEY,
  video_id BIGINT NOT NULL,
  user_id VARCHAR(50),
  text VARCHAR(255),
  time_in_video DOUBLE,
  send_time DATETIME
);

☁️ Redis 数据结构设计

☁️ Redis 存弹幕（实时 + 历史）

• 弹幕按 timeInVideo 入 Redis List
• 前端加载 Redis 弹幕，根据视频播放进度展示
• 每隔 30 秒自动将 Redis 弹幕落库并清除缓存

🔐 敏感词过滤系统（服务 + 接口）

🔧 Redis Filter Service

@Service
public class DanmuFilterService {
    @Autowired RedisTemplate<String, String> redis;
    public boolean containsForbidden(String text) {
        Set<String> words = redis.opsForSet().members("filter:words");
        return words != null && words.stream().anyMatch(text::contains);
    }
}

🔧 管理接口

@RestController
@RequestMapping("/api/filters")
public class FilterController {

    @Autowired RedisTemplate<String, String> redis;

    @PostMapping("/add")
    public String add(@RequestParam String word) {
        redis.opsForSet().add("filter:words", word);
        return "添加成功";
    }

    @PostMapping("/remove")
    public String remove(@RequestParam String word) {
        redis.opsForSet().remove("filter:words", word);
        return "删除成功";
    }

    @GetMapping("/list")
    public Set<String> list() {
        return redis.opsForSet().members("filter:words");
    }
}

🚦 弹幕限频控制

👮 Redis 限流器

@Service
public class DanmuRateLimitService {
    @Autowired RedisTemplate<String, String> redis;

    public boolean isTooFast(String userId) {
        String key = "limit:user:" + userId;
        if (redis.hasKey(key)) return true;
        redis.opsForValue().set(key, "1", Duration.ofSeconds(2));
        return false;
    }
}

🔄 定时将弹幕持久化到 MySQL

@Component
public class DanmuBackupTask {

    @Autowired RedisTemplate<String, String> redis;
    @Autowired DanmuRepository danmuRepo;
    Gson gson = new Gson();

    @Scheduled(fixedRate = 30000) // 每 30 秒
    public void flushToDb() {
        Set<String> keys = redis.keys("danmu:video:*");
        if (keys == null) return;

        for (String key : keys) {
            List<String> list = redis.opsForList().range(key, 0, -1);
            if (list == null || list.isEmpty()) continue;
            List<Danmu> danmus = list.stream().map(j -> gson.fromJson(j, Danmu.class)).toList();
            danmuRepo.saveAll(danmus);
            redis.delete(key); // 清空 Redis
        }
    }
}

📡 WebSocket 处理器（敏感词 + 限频 + 广播）

@ServerEndpoint("/ws/danmu/{videoId}/{userId}")
@Component
public class DanmuWebSocket {

    private static final Map<String, Session> sessions = new ConcurrentHashMap<>();

    private static DanmuFilterService filterService;
    private static DanmuRateLimitService rateLimitService;
    private static RedisTemplate<String, String> redis;

    @Autowired
    public void setDeps(DanmuFilterService f, DanmuRateLimitService r, RedisTemplate<String, String> rt) {
        filterService = f;
        rateLimitService = r;
        redis = rt;
    }

    @OnOpen
    public void onOpen(Session session) {
        sessions.put(session.getId(), session);
    }

    @OnMessage
    public void onMessage(String msgJson, Session session,
                          @PathParam("videoId") String videoId,
                          @PathParam("userId") String userId) {
        Danmu danmu = new Gson().fromJson(msgJson, Danmu.class);
        danmu.setUserId(userId);
        danmu.setSendTime(LocalDateTime.now());

        // 限频
        if (rateLimitService.isTooFast(userId)) {
            sendTo(session, "[系统通知] 请勿频繁发送弹幕！");
            return;
        }

        // 敏感词
        if (filterService.containsForbidden(danmu.getText())) {
            sendTo(session, "[系统通知] 弹幕含违禁词，已屏蔽！");
            return;
        }

        // 存 Redis
        redis.opsForList().rightPush("danmu:video:" + videoId, new Gson().toJson(danmu));

        // 广播
        sessions.values().forEach(s -> sendTo(s, new Gson().toJson(danmu)));
    }

    private void sendTo(Session session, String msg) {
        try { session.getBasicRemote().sendText(msg); } catch (Exception e) {}
    }

    @OnClose
    public void onClose(Session session) {
        sessions.remove(session.getId());
    }
}

💻 前端弹幕逻辑（伪代码）

// 加载历史弹幕
fetch("/api/danmu/history?videoId=123")
  .then(res => res.json())
  .then(data => {
    danmus = data.sort((a, b) => a.time - b.time);
  });

setInterval(() => {
  const currentTime = video.currentTime;
  while (danmus.length && danmus[0].time <= currentTime) {
    showDanmu(danmus.shift().text);
  }
}, 200);

// 连接 WebSocket
const ws = new WebSocket("ws://localhost:8080/ws/danmu/123/userA");
ws.onmessage = e => showDanmu(JSON.parse(e.data).text);

// 发送弹幕
function sendDanmu(text) {
  ws.send(JSON.stringify({ text, time: video.currentTime }));
}

✅ 最终效果

🧪 当前系统存在的缺点分析

✅ 总结建议