SpringBoot + RabbitMQ + Redis + MySQL：社交平台私信发送、已读状态同步与历史消息缓存

2501_94505843

942人浏览 · 2026-01-09 13:41:25

2501_94505843 · 2026-01-09 13:41:25 发布

SpringBoot + RabbitMQ + Redis + MySQL：社交平台私信发送、已读状态同步与历史消息缓存

今天就结合最新项目，聊聊如何用 SpringBoot+RabbitMQ+Redis+MySQL 打造一套稳定、高效的私信系统，涵盖消息发送、已读状态同步和历史消息缓存三大核心场景。内容会穿插这些年积累的实战经验，拒绝空谈理论。

一、为什么是这套技术组合？从业务痛点倒推

做私信系统前，先得想清楚业务上最痛的点是什么。八年经验告诉我，用户对私信的核心诉求就三个：不丢消息、状态实时、查得快。对应到技术上，就是可靠性、一致性和性能。

我们先看一下主流技术组合的对比：

方案	优势	短板	适合场景
纯 MySQL	实现简单	高并发下写入慢，已读状态更新锁冲突	日活 10 万以下的小平台
Redis+MySQL	读快，状态更新方便	消息发送同步阻塞，峰值易崩	中等流量，但消息发送不频繁
Kafka+MySQL	吞吐量极高	架构重，已读状态同步麻烦	类似微博的 "私信广播" 场景
RabbitMQ+Redis+MySQL	异步可靠，状态实时，读写分离	组件多，运维成本略高	社交平台私信（核心需求：可靠 + 实时 + 快速查询）

最终选择SpringBoot+RabbitMQ+Redis+MySQL，正是因为它能完美解决三个核心痛点：

RabbitMQ：确保消息不丢（Confirm 机制 + 死信队列），异步发送不阻塞主线程
Redis：毫秒级更新已读状态（Hash 结构），缓存最近消息（List+ZSet）
MySQL：持久化存储历史消息，支持复杂查询（如按时间范围、按用户）
SpringBoot：快速整合上述组件，减少 boilerplate 代码

可能有人会问："为什么不用 Kafka 替代 RabbitMQ？"—— 在私信场景中，消息通常是点对点的，Kafka 的 "广播" 特性用不上，反而 RabbitMQ 的交换机类型（Direct/Topic）更适合私信的路由规则，而且单条消息确认机制更灵活。

二、整体架构设计：从 "发送" 到 "读取" 的全链路

先上一张架构流程图，让大家对整个流程有个直观认识：

css

体验AI代码助手

代码解读

复制代码

A[用户A发送私信] -->|1. 调用API| B[SpringBoot应用] B -->|2. 校验权限| C{权限校验} C -->|不通过| D[返回403] C -->|通过| E[生成消息ID+雪花算法] E -->|3. 发送到RabbitMQ| F[私信交换机(direct)] F -->|4. 路由到队列| G[用户B的私信队列] B -->|5. 先写Redis缓存| H[Redis: 未读消息List + 已读状态Hash] G -->|6. 消费消息| I[消息消费者] I -->|7. 持久化到MySQL| J[message表] I -->|8. 更新Redis计数| K[未读消息数自增] L[用户B查看私信] -->|9. 读取缓存| H L -->|10. 标记已读| M[更新Redis Hash + 发送已读确认到RabbitMQ] M -->|11. 异步更新MySQL| J N[加载历史消息] -->|12. 优先查Redis| O[缓存命中] N -->|13. 缓存未命中| P[查MySQL + 回填Redis]

整个流程分为三个核心链路：

消息发送链路：客户端→API→权限校验→RabbitMQ 异步写入→Redis 临时存储→MySQL 持久化
已读状态同步链路：读消息→Redis 标记已读→RabbitMQ 异步通知→MySQL 更新
历史消息查询链路：优先查 Redis 缓存→未命中则查 MySQL→结果回填 Redis

这种设计的好处是：

发送消息时用 RabbitMQ 异步处理，避免用户等待
已读状态先更 Redis 保证实时性，再异步同步到 MySQL
热点数据（最近消息）放 Redis，冷数据放 MySQL，平衡性能和存储成本

三、核心功能实现：代码 + 经验，干货满满

3.1 第一步：消息发送 —— 如何保证 "不丢消息"

私信系统最忌讳的就是 "消息丢失"，用户 A 发了消息，用户 B 收不到，这是致命问题。我们用 RabbitMQ 的 Confirm 机制 + 死信队列来解决。

3.1.1 RabbitMQ 配置

typescript

体验AI代码助手

代码解读

复制代码

@Configuration public class RabbitMqConfig { // 私信交换机（direct类型，点对点路由） public static final String MSG_EXCHANGE = "private_msg_exchange"; // 死信交换机 public static final String DEAD_LETTER_EXCHANGE = "msg_dead_letter_exchange"; // 声明交换机 @Bean public DirectExchange msgExchange() { // 持久化，不自动删除 return ExchangeBuilder.directExchange(MSG_EXCHANGE).durable(true).build(); } @Bean public DirectExchange deadLetterExchange() { return ExchangeBuilder.directExchange(DEAD_LETTER_EXCHANGE).durable(true).build(); } // 动态创建用户私信队列（每个用户一个队列，队列名：private_msg_queue_{userId}） public Queue getUserMsgQueue(Long userId) { // 配置死信队列参数 Map<String, Object> args = new HashMap<>(); // 消息过期时间：10分钟（10*60*1000ms） args.put("x-message-ttl", 600000); // 死信交换机 args.put("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE); // 死信路由键（用户ID） args.put("x-dead-letter-routing-key", "dead_" + userId); // 队列名：private_msg_queue_10086 return QueueBuilder.durable("private_msg_queue_" + userId) .withArguments(args) .build(); } // 绑定队列到交换机（路由键为接收者用户ID） @Bean public Binding bindUserQueue(Long userId) { return BindingBuilder.bind(getUserMsgQueue(userId)) .to(msgExchange()) .with(String.valueOf(userId)); } // 死信队列（处理发送失败的消息） @Bean public Queue deadLetterQueue() { return QueueBuilder.durable("msg_dead_letter_queue").build(); } // 绑定死信队列 @Bean public Binding bindDeadLetterQueue() { return BindingBuilder.bind(deadLetterQueue()) .to(deadLetterExchange()) .with("dead_*"); // 匹配所有死信路由键 } // 消息确认配置 @Bean public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) { RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory); // 开启Confirm机制（确认消息是否到达交换机） rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> { if (!ack) { log.error("消息发送到交换机失败，原因：{}", cause); // 失败处理：记录到数据库，后续重试 if (correlationData != null) { String msgId = correlationData.getId(); messageRetryService.recordFailedMsg(msgId, cause); } } }); // 开启Return机制（确认消息是否到达队列） rabbitTemplate.setReturnsCallback(returnedMessage -> { log.error("消息未到达队列，消息体：{}，原因：{}", new String(returnedMessage.getMessage().getBody()), returnedMessage.getReplyText()); // 处理逻辑同上 }); return rabbitTemplate; } }

3.1.2 消息发送服务

typescript

体验AI代码助手

代码解读

复制代码

@Service @Slf4j public class MessageSendService { @Autowired private RabbitTemplate rabbitTemplate; @Autowired private RedisTemplate<String, Object> redisTemplate; @Autowired private MessageMapper messageMapper; // 消息存储的Redis键：未读消息列表（每个用户一个List） private static final String UNREAD_MSG_KEY = "private:msg:unread:%s"; // %s为接收者ID /** * 发送私信 */ @Transactional public void sendMessage(MessageDTO dto) { // 1. 生成唯一消息ID（雪花算法） Long msgId = SnowflakeIdGenerator.generateId(); // 2. 构建消息实体 Message message = new Message(); message.setId(msgId); message.setSenderId(dto.getSenderId()); message.setReceiverId(dto.getReceiverId()); message.setContent(dto.getContent()); message.setSendTime(LocalDateTime.now()); message.setReadStatus(0); // 0-未读，1-已读 try { // 3. 先存Redis（保证接收方快速看到新消息） String unreadKey = String.format(UNREAD_MSG_KEY, dto.getReceiverId()); redisTemplate.opsForList().leftPush(unreadKey, message); // 设置过期时间：7天（超过7天未读也会被持久化） redisTemplate.expire(unreadKey, 7, TimeUnit.DAYS); // 4. 发送到RabbitMQ（异步持久化到MySQL） // 构建消息属性，设置correlationId用于确认 CorrelationData correlationData = new CorrelationData(msgId.toString()); // 路由键为接收者ID，确保消息进入正确队列 rabbitTemplate.convertAndSend( RabbitMqConfig.MSG_EXCHANGE, String.valueOf(dto.getReceiverId()), message, correlationData ); log.info("消息发送成功，msgId:{}", msgId); } catch (Exception e) { log.error("消息发送失败", e); // 极端情况：Redis和MQ都挂了，直接写MySQL保证不丢 messageMapper.insert(message); throw new BusinessException("消息发送失败，请重试"); } } }

3.1.3 消息消费服务（持久化到 MySQL）

less

体验AI代码助手

代码解读

复制代码

@Component @Slf4j public class MessageConsumer { @Autowired private MessageMapper messageMapper; @Autowired private RedisTemplate<String, Object> redisTemplate; // 未读消息计数的Redis键 private static final String UNREAD_COUNT_KEY = "private:msg:unread:count:%s"; // %s为接收者ID /** * 消费消息，持久化到MySQL */ @RabbitListener(queuesToDeclare = @Queue( value = "#{T(com.example.config.RabbitMqConfig).getUserMsgQueue(10086).getName()}", durable = "true" )) public void consumeMessage(Message message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException { try { // 1. 持久化到MySQL messageMapper.insert(message); // 2. 更新未读消息计数 String countKey = String.format(UNREAD_COUNT_KEY, message.getReceiverId()); redisTemplate.opsForValue().increment(countKey); // 3. 手动确认消息 channel.basicAck(tag, false); log.info("消息持久化成功，msgId:{}", message.getId()); } catch (Exception e) { log.error("消息消费失败，msgId:{}", message.getId(), e); // 消费失败，拒绝消息并放回队列（最多重试3次） if (getRetryCount(message) < 3) { channel.basicNack(tag, false, true); } else { // 超过重试次数，放入死信队列 channel.basicNack(tag, false, false); } } } // 获取消息重试次数（从消息属性中获取） private int getRetryCount(Message message) { // 实际项目中可通过消息属性记录重试次数 return 0; } }

八年经验总结：

消息 ID 一定要用雪花算法，确保全局唯一，方便追踪和去重
必须开启 RabbitMQ 的手动确认模式，避免消费者崩溃导致消息丢失
极端情况下（Redis 和 MQ 都挂了），要能降级到直接写 MySQL，这是最后一道防线

3.2 第二步：已读状态同步 —— 如何保证 "实时一致"

已读状态不同步是用户吐槽最多的点："我明明读了，对方还显示未读" 或者 "我没读，却显示已读"。我们用 Redis 的 Hash 结构存储实时状态，配合 RabbitMQ 异步同步到 MySQL。

3.2.1 已读状态服务

ini

体验AI代码助手

代码解读

复制代码

@Service public class ReadStatusService { @Autowired private RedisTemplate<String, Object> redisTemplate; @Autowired private RabbitTemplate rabbitTemplate; @Autowired private MessageMapper messageMapper; // 已读状态的Redis键：Hash结构，key=msgId，value=readStatus private static final String READ_STATUS_KEY = "private:msg:read:status:%s"; // %s为接收者ID // 已读确认交换机 private static final String READ_CONFIRM_EXCHANGE = "read_confirm_exchange"; /** * 标记消息为已读 */ public void markAsRead(Long receiverId, List<Long> msgIds) { if (CollectionUtils.isEmpty(msgIds)) { return; } String statusKey = String.format(READ_STATUS_KEY, receiverId); // 1. 批量更新Redis中的已读状态 Map<String, Integer> statusMap = new HashMap<>(); for (Long msgId : msgIds) { statusMap.put(msgId.toString(), 1); // 1表示已读 } redisTemplate.opsForHash().putAll(statusKey, statusMap); // 2. 从Redis未读列表中移除这些消息 String unreadKey = String.format(MessageSendService.UNREAD_MSG_KEY, receiverId); msgIds.forEach(msgId -> redisTemplate.opsForList().remove(unreadKey, 0, msgId)); // 3. 更新未读计数（减少对应数量） String countKey = String.format(MessageConsumer.UNREAD_COUNT_KEY, receiverId); redisTemplate.opsForValue().decrement(countKey, msgIds.size()); // 4. 发送已读确认到RabbitMQ，异步更新MySQL ReadConfirmDTO confirmDTO = new ReadConfirmDTO(); confirmDTO.setReceiverId(receiverId); confirmDTO.setMsgIds(msgIds); confirmDTO.setReadTime(LocalDateTime.now()); rabbitTemplate.convertAndSend(READ_CONFIRM_EXCHANGE, "read_confirm", confirmDTO); } /** * 获取消息的已读状态 */ public Map<Long, Integer> getReadStatus(Long receiverId, List<Long> msgIds) { String statusKey = String.format(READ_STATUS_KEY, receiverId); List<Object> values = redisTemplate.opsForHash().multiGet( statusKey, msgIds.stream().map(String::valueOf).collect(Collectors.toList()) ); Map<Long, Integer> result = new HashMap<>(); for (int i = 0; i < msgIds.size(); i++) { Long msgId = msgIds.get(i); Object value = values.get(i); // Redis中没有则查MySQL（缓存未命中） if (value == null) { Integer status = messageMapper.selectReadStatusById(msgId, receiverId); result.put(msgId, status); // 回填Redis redisTemplate.opsForHash().put(statusKey, msgId.toString(), status); } else { result.put(msgId, (Integer) value); } } return result; } /** * 消费已读确认，更新MySQL */ @RabbitListener(queues = "read_confirm_queue") public void handleReadConfirm(ReadConfirmDTO dto) { messageMapper.batchUpdateReadStatus(dto.getReceiverId(), dto.getMsgIds(), dto.getReadTime()); log.info("批量更新已读状态成功，接收者:{}，消息数:{}", dto.getReceiverId(), dto.getMsgIds().size()); } }

实战技巧：

已读状态用 Redis Hash 存储，key 是接收者 ID，field 是消息 ID，value 是状态，查询效率极高
批量标记已读时，先更 Redis 保证实时性，再异步同步 MySQL，用户体验更好
缓存未命中时，查 MySQL 后一定要回填 Redis，避免重复查询数据库

3.3 第三步：历史消息缓存 —— 如何做到 "查得快"

用户查看历史消息时，翻页加载不能卡，这就需要合理的缓存策略。我们把最近 30 天的消息放 Redis，更早的放 MySQL，同时用 ZSet 做时间范围查询。

3.3.1 历史消息服务

ini

体验AI代码助手

代码解读

复制代码

@Service public class HistoryMessageService { @Autowired private RedisTemplate<String, Object> redisTemplate; @Autowired private MessageMapper messageMapper; // 历史消息缓存的Redis键：ZSet结构，score=时间戳，value=消息JSON private static final String HISTORY_MSG_KEY = "private:msg:history:%s:%s"; // %s为发送者ID:%s为接收者ID // 缓存时间：30天 private static final long CACHE_DAYS = 30; /** * 加载历史消息（分页） */ public PageResult<Message> loadHistoryMessages(Long senderId, Long receiverId, LocalDateTime startTime, LocalDateTime endTime, int pageNum, int pageSize) { // 构建Redis键（确保senderId < receiverId，避免重复存储） Long minId = Math.min(senderId, receiverId); Long maxId = Math.max(senderId, receiverId); String historyKey = String.format(HISTORY_MSG_KEY, minId, maxId); // 1. 先查Redis缓存 Set<ZSetOperations.TypedTuple<Object>> tuples = null; try { long startScore = startTime.toEpochMilli(); long endScore = endTime.toEpochMilli(); // ZSet查询：按时间范围[startScore, endScore]，分页 int start = (pageNum - 1) * pageSize; int end = pageNum * pageSize - 1; tuples = redisTemplate.opsForZSet().rangeByScoreWithScores( historyKey, startScore, endScore, start, pageSize ); } catch (Exception e) { log.warn("Redis查询历史消息失败， fallback到MySQL", e); } // 2. 缓存命中，解析结果 if (tuples != null && !tuples.isEmpty()) { List<Message> messages = tuples.stream() .map(tuple -> JSON.parseObject(tuple.getValue().toString(), Message.class)) .sorted(Comparator.comparing(Message::getSendTime)) // 按时间排序 .collect(Collectors.toList()); // 查询总数（用于分页） Long total = redisTemplate.opsForZSet().count(historyKey, startTime.toEpochMilli(), endTime.toEpochMilli()); return new PageResult<>(messages, total, pageNum, pageSize); } // 3. 缓存未命中，查MySQL Page<Message> dbPage = messageMapper.queryHistoryMessages( senderId, receiverId, startTime, endTime, PageHelper.startPage(pageNum, pageSize) ); // 4. 回填Redis（异步，不阻塞当前请求） CompletableFuture.runAsync(() -> { try { ZSetOperations<String, Object> zSetOps = redisTemplate.opsForZSet(); for (Message msg : dbPage.getResult()) { // 只缓存30天内的消息 if (ChronoUnit.DAYS.between(msg.getSendTime(), LocalDateTime.now()) <= CACHE_DAYS) { zSetOps.add(historyKey, JSON.toJSONString(msg), msg.getSendTime().toEpochMilli()); } } // 设置过期时间 redisTemplate.expire(historyKey, CACHE_DAYS, TimeUnit.DAYS); } catch (Exception e) { log.error("回填历史消息到Redis失败", e); } }); return new PageResult<>(dbPage.getResult(), dbPage.getTotal(), pageNum, pageSize); } }

性能优化点：

缓存键设计成minId:maxId，避免 sender 和 receiver 互换导致的重复存储（比如 A→B 和 B→A 的消息存在同一个键下）
用 ZSet 存储历史消息，score 是时间戳，天然支持按时间范围查询和分页
回填 Redis 用异步线程，不阻塞用户查询
只缓存最近 30 天的消息，平衡缓存大小和查询效率

四、踩坑实录：这些坑我替你踩过了

4.1 坑 1：消息重复消费导致 MySQL 中出现重复数据

问题：早期用了 RabbitMQ 的自动确认模式，消费者处理到一半宕机，重启后消息被重新消费，导致同一条消息在 MySQL 中存了多条。

解决方案：

改用手动确认模式，处理完成后再 ack
在 MySQL 的 message 表中给 msgId 加唯一索引，即使重复消费也会入库失败
代码层面做幂等性校验：消费前先查 msgId 是否已存在

sql

体验AI代码助手

代码解读

复制代码

-- 给msgId加唯一索引 ALTER TABLE `message` ADD UNIQUE INDEX `idx_msg_id` (`id`);

4.2 坑 2：Redis 缓存雪崩，历史消息查询全部打在 MySQL 上

问题：某天 Redis 集群升级重启，所有缓存失效，大量用户同时查历史消息，MySQL 瞬间被打满，出现连接超时。

解决方案：

给 Redis 缓存设置随机过期时间（比如 30±1 天），避免缓存同时失效
MySQL 层面加读写分离，历史消息查询走从库
加熔断降级：当 MySQL 压力过大时，返回 "加载中，请稍后重试"

scss

体验AI代码助手

代码解读

复制代码

// 设置随机过期时间 int randomDays = new Random().nextInt(2) + 30; // 30-31天 redisTemplate.expire(historyKey, randomDays, TimeUnit.DAYS);

4.3 坑 3：已读状态更新后，未读计数没同步

问题：用户标记已读后，未读计数偶尔不减少，需要刷新才显示正确。

原因：高并发下，decrement操作和remove操作有先后顺序问题，导致计数不准。

解决方案：

用 Redis 事务保证操作原子性
定期（如每天凌晨）全量同步未读计数到 Redis

scss

体验AI代码助手

代码解读

复制代码

// 用事务保证原子性 redisTemplate.execute(new SessionCallback<Object>() { @Override public Object execute(RedisOperations operations) throws DataAccessException { operations.multi(); // 移除未读消息 msgIds.forEach(msgId -> operations.opsForList().remove(unreadKey, 0, msgId)); // 减少计数 operations.opsForValue().decrement(countKey, msgIds.size()); return operations.exec(); } });