基于SpringBoot+RabbitMQ+Redis的私信系统架构设计与实战-育师

私信系统这东西，看着功能简单，不就是"你发一条、我收一条"嘛。可真要在一个日活几十万的社交平台上落地，从消息不丢、不乱序，到已读回执实时同步，再到历史消息秒开不卡顿，每一环都是坑。我去年完整做了一版基于 SpringBoot + RabbitMQ + Redis + MySQL 的私信模块，从设计文档到上线压测都跑了一遍，这篇就把整套方案、核心代码思路和踩过的坑一起梳理出来，给正在设计 IM 或站内信的同学做个参考。

1. 整体架构设计与核心思路拆解

先讲清楚为什么选这套技术栈。私信系统的核心诉求有三个：发送要快、状态要准、历史要全。这三个诉求单靠任何一款中间件都搞不定，必须各司其职。

1.1 技术选型背后的取舍逻辑

MySQL 作为最终数据落库的"唯一真相源"，存全量消息和历史记录。RabbitMQ 负责异步削峰和解耦，用户点发送按钮后接口立刻返回，真正的推送和落库动作放到消息队列里慢慢消化。Redis 则承担两个职责：一是热点消息的缓存，让用户翻聊天记录时不用每次都查 MySQL；二是已读状态和未读计数的瞬时存储，因为已读状态是高频写操作，直接写 MySQL 会把数据库拖垮。

这套组合的本质是把"写路径"和"读路径"拆开。写路径走 MySQL + RabbitMQ，保证可靠性和顺序性；读路径走 Redis，保证性能和体验。刚开始我也纠结过要不要直接用 WebSocket 长连接做实时推送，但私信场景和群聊不一样，用户在线状态不稳定，离线消息必须要靠消息队列兜底，RabbitMQ 的持久化机制天然适合干这个活。

1.2 整体数据流向全景图

我习惯先把链路图画清楚再动手写代码。整个私信系统的主流程是这样的：

发送方调用POST /api/im/message/send接口，请求体包含接收方 uid、消息类型（文本/图片/语音）、消息内容。
SpringBoot 接口层做参数校验后，先把消息主记录写入 MySQL 的im_message表，此时消息状态是SENDING。
同步构造消息 ID（用雪花算法生成），把消息内容序列化后投递到 RabbitMQ 的im.message.queue队列。
接口立刻返回"发送成功"给客户端，真正的投递动作由异步消费者完成。
消费者从队列拿到消息后，依次执行三个动作：更新 MySQL 消息状态为SENT、写入 Redis 对应会话的 ZSET 缓存、通过 WebSocket 推送在线通知给接收方。
接收方打开聊天窗口时，客户端先拉 Redis 缓存的最近消息，再调已读接口上报最后一条已读消息 ID。

这套流程的关键在于：接口层和消息投递层完全解耦。即使 RabbitMQ 短暂不可用，消息也已经落进了 MySQL，等 MQ 恢复后消费端还能从队列继续处理，不会出现"用户以为发出去了、实际丢消息"的严重事故。

2. 私信发送链路：从接口到队列的完整实现

发送链路是整个系统的入口，也是可靠性要求最高的部分。我见过不少团队把消息先发 MQ 再落库，顺序搞反了，结果 MQ 一抖动消息就丢了。正确的做法是"先落库，再投递"，保证至少一次投递。

2.1 消息表结构设计与索引优化

消息表是私信系统的核心表，设计时我重点考虑了三个问题：查询效率、分页游标、消息状态流转。

CREATE TABLE `im_message` ( `id` bigint(20) NOT NULL COMMENT '雪花算法生成的消息ID', `conversation_id` varchar(64) NOT NULL COMMENT '会话ID，由双方uid拼接生成', `from_uid` bigint(20) NOT NULL COMMENT '发送方用户ID', `to_uid` bigint(20) NOT NULL COMMENT '接收方用户ID', `content` text NOT NULL COMMENT '消息内容', `msg_type` tinyint(4) NOT NULL DEFAULT '1' COMMENT '消息类型：1文本 2图片 3语音', `status` tinyint(4) NOT NULL DEFAULT '0' COMMENT '状态：0发送中 1已发送 2已读 3撤回', `create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (`id`), KEY `idx_conversation_time` (`conversation_id`, `create_time`), KEY `idx_from_uid` (`from_uid`), KEY `idx_to_uid` (`to_uid`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='私信消息表';

这里有两个细节必须强调。第一个是conversation_id的生成规则，我采用的是min(from_uid, to_uid)_max(from_uid, to_uid)这种对称拼接方式，这样不管从哪个方向发消息，查会话时都能命中同一个 ID，避免查历史消息时需要(from_uid = A AND to_uid = B) OR (from_uid = B AND to_uid = A)这种烂索引。

第二个是查询历史消息时不要用 OFFSET 分页，而是用"游标分页"。因为私信消息是持续增长的，用户翻页过程中如果来了新消息，OFFSET 分页会导致重复数据或跳数据。游标分页只需要记住最后一条消息的 create_time（或消息ID），下次查询用WHERE conversation_id = ? AND create_time < ? ORDER BY create_time DESC LIMIT 20就能稳定翻页。

2.2 RabbitMQ 交换机与队列配置要点

RabbitMQ 的配置直接影响消息投递的可靠性。我用的方案是直连交换机（Direct Exchange）+ 持久化队列 + 手动 ACK。

@Configuration public class RabbitMQConfig { public static final String IM_EXCHANGE = "im.exchange"; public static final String IM_MESSAGE_QUEUE = "im.message.queue"; public static final String IM_MESSAGE_ROUTING_KEY = "im.message.send"; public static final String IM_DLX_EXCHANGE = "im.dlx.exchange"; public static final String IM_DLX_QUEUE = "im.dlx.queue"; public static final String IM_DLX_ROUTING_KEY = "im.dlx.routing"; @Bean public DirectExchange imExchange() { return new DirectExchange(IM_EXCHANGE, true, false); } @Bean public Queue imMessageQueue() { Map<String, Object> args = new HashMap<>(); // 设置死信交换机 args.put("x-dead-letter-exchange", IM_DLX_EXCHANGE); args.put("x-dead-letter-routing-key", IM_DLX_ROUTING_KEY); // 设置队列最大长度，防止消息积压撑爆内存 args.put("x-max-length", 100000); return new Queue(IM_MESSAGE_QUEUE, true, false, false, args); } @Bean public Binding imBinding() { return BindingBuilder.bind(imMessageQueue()) .to(imExchange()) .with(IM_MESSAGE_ROUTING_KEY); } @Bean public DirectExchange imDlxExchange() { return new DirectExchange(IM_DLX_EXCHANGE, true, false); } @Bean public Queue imDlxQueue() { return new Queue(IM_DLX_QUEUE, true); } @Bean public Binding imDlxBinding() { return BindingBuilder.bind(imDlxQueue()) .to(imDlxExchange()) .with(IM_DLX_ROUTING_KEY); } }

这里最值得说的是死信队列（DLX）。我一开始没配死信，结果消费者代码有 bug 时，消息一直在队列里被重新投递，形成死循环。配置了死信交换机后，消费失败的消息会自动转移到im.dlx.queue，不会阻塞主队列，排查问题时也能直接在死信队列里看到失败原因。

消费者这边，我强烈建议用手动 ACK 模式，配合basicNack的 requeue 参数来控制消息去向。

@Component @Slf4j public class MessageConsumer { @RabbitListener(queues = RabbitMQConfig.IM_MESSAGE_QUEUE, ackMode = "MANUAL") public void onMessage(Message message, Channel channel) throws IOException { long deliveryTag = message.getMessageProperties().getDeliveryTag(); try { // 1. 反序列化消息体 // 2. 更新MySQL消息状态为已发送 // 3. 写入Redis会话缓存 // 4. 推送WebSocket在线通知 channel.basicAck(deliveryTag, false); } catch (Exception e) { log.error("消息消费失败: {}", e.getMessage(), e); // 业务异常：记录日志后确认消息，避免死循环 channel.basicNack(deliveryTag, false, false); } } }

关于basicNack的第三个参数requeue，我踩过一个大坑。最初设置为true，意思是消费失败重新入队。但如果是消息本身的数据问题（比如内容超长导致反序列化失败），重新入队一百次也还是失败，反而把 CPU 打满。后来我改成false，配合死信交换机，把异常消息转储到 DLX 队列，人工排查后补发。

2.3 发送接口幂等性设计

私信发送还有一个容易被忽略的问题：接口幂等。客户端网络抖动导致用户点了两次发送按钮，如果后端不做幂等处理，接收方就会收到两条一模一样的消息。

我的方案是让客户端在请求头传一个clientMsgId（客户端生成的消息唯一标识），后端在处理写入 MySQL 之前先查一下im_message表里是否已存在相同client_msg_id的消息。为了这个查询高效，需要在消息表加一个client_msg_id字段并建唯一索引。

不光是发送接口，已读上报接口同样需要幂等设计。用户频繁打开关闭聊天窗口，已读接口可能被调用几十次，每次都更新数据库完全没必要。后面讲已读状态时会详细说怎么用 Redis 缓冲来解决这个问题。

3. 已读状态同步：Redis 缓冲 + MySQL 异步落库

已读状态是私信系统里最容易被低估的技术点。表面上看就是一个布尔值：已读或者未读。但真要做得"实时且不拖垮数据库"，就需要仔细设计读写路径。

3.1 为什么不能直接写 MySQL

假设一个用户有 200 个会话，每个会话有一条未读消息。用户打开 App 后，客户端会并发上报 200 个已读请求。如果每个请求都直接UPDATE im_message SET status = 2 WHERE id = ?，数据库瞬间要处理 200 条 update，而且这些 update 还都加了行锁。在高峰期，大量用户的已读上报会让 InnoDB 的锁竞争变得非常激烈。

更麻烦的是，已读接口的调用频率远高于发送接口。用户每打开一次聊天窗口就会触发一次，发送一条消息只会触发一次。读多写多的场景，必须用 Redis 做一层缓冲。

3.2 Redis 已读状态存储方案

我采用的方案是：已读状态第一优先写 Redis，然后通过异步批处理回写 MySQL。

Redis 里维护两个维度的数据：

会话维度已读位置：im:read:{conversation_id}:{user_id}，存储用户在该会话中已读到的最大消息 ID。用 String 类型即可。
未读计数：im:unread:{user_id}:{conversation_id}，存储用户在某会话的未读消息数。

已读上报接口的逻辑很简单：

@Service public class ReadStatusService { @Autowired private StringRedisTemplate redisTemplate; /** * 上报已读状态 */ public void markAsRead(Long userId, String conversationId, Long lastReadMsgId) { String readKey = "im:read:" + conversationId + ":" + userId; // 先读Redis里的旧值，比对后决定是否更新 String oldValue = redisTemplate.opsForValue().get(readKey); if (oldValue != null && Long.parseLong(oldValue) >= lastReadMsgId) { // 已读位置没有前移，直接忽略 return; } redisTemplate.opsForValue().set(readKey, String.valueOf(lastReadMsgId)); // 清除未读计数 String unreadKey = "im:unread:" + userId + ":" + conversationId; redisTemplate.delete(unreadKey); // 把本次已读事件放入异步队列，等待批量落库 asyncBatchSaveReadRecord(userId, conversationId, lastReadMsgId); } }

这段代码的精髓在于先用 Redis 做了"去重判断"。用户连续打开同一个会话十次，只有第一次会真正触发后续逻辑，后九次因为oldValue >= lastReadMsgId直接返回，连异步落库的请求都不会发。

3.3 已读状态批量落库策略

异步落库我用的是"定时批量合并"策略，而不是每条已读事件都触发一次数据库更新。具体做法是：把已读事件放进一个内存队列或 Redis 列表，定时任务每 3 秒合并一次，同一个会话同一个用户只保留最大的已读消息 ID，然后批量 UPDATE。

@Component public class ReadRecordBatchTask { @Scheduled(fixedDelay = 3000) public void flushReadRecords() { // 1. 从队列取出所有待落库的已读记录 // 2. 按 conversation_id + user_id 分组，每组取最大 lastReadMsgId // 3. 批量执行 UPDATE im_read_record SET last_read_msg_id = ? // WHERE user_id = ? AND conversation_id = ? // 4. 同时更新 im_message 表：把该会话内小于等于 lastReadMsgId // 且 status = 1 的消息批量置为 2（已读） } }

这里要特别说明 MySQL 侧的已读记录表设计。私信已读状态不建议直接在消息表上更新每一行的 status 字段，因为高并发下会产生大量行锁。我单独建了一张已读记录表，记录用户在每个会话中的"已读位置"。

CREATE TABLE `im_read_record` ( `id` bigint(20) NOT NULL AUTO_INCREMENT, `user_id` bigint(20) NOT NULL COMMENT '用户ID', `conversation_id` varchar(64) NOT NULL COMMENT '会话ID', `last_read_msg_id` bigint(20) NOT NULL COMMENT '最后已读消息ID', `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (`id`), UNIQUE KEY `uk_user_conv` (`user_id`, `conversation_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='已读记录表';

用INSERT ... ON DUPLICATE KEY UPDATE实现"有则更新、无则插入"，配合唯一索引uk_user_conv，即使多个线程同时写也安全。这张表在查询"某用户所有会话的未读数"时也很有用，可以直接 JOIN 会话表统计。

4. 历史消息缓存：Redis 如何扛住高频读取

私信的历史消息读取是最典型的读多写少场景。用户翻聊天记录时，每次都查 MySQL 显然不行。Redis 的 ZSET（有序集合）是为这个场景量身定做的数据结构。

4.1 基于 ZSET 的会话消息缓存设计

我用的缓存方案是"双 Key"结构：

会话消息索引：im:conv:{conversation_id}，类型为 ZSET，member 存消息 ID，score 存消息时间戳。
消息内容：im:msg:{message_id}，类型为 String，存放消息内容 JSON。

为什么要把索引和内容拆开？因为用户翻聊天记录时，不一定每条消息的内容都需要立即展示。先通过 ZSET 拿到一批消息 ID，再按需去取消息内容，万一某条消息内容已经从缓存中淘汰了，还可以回源 MySQL 查，不会导致整页数据缺失。

写入缓存的逻辑在消息消费者中完成：

public void cacheMessage(ImMessageDTO message) { String convKey = "im:conv:" + message.getConversationId(); String msgKey = "im:msg:" + message.getMessageId(); // 写入消息内容，设置5天过期 redisTemplate.opsForValue().set(msgKey, JSON.toJSONString(message), 5, TimeUnit.DAYS); // 消息ID加入会话ZSET，score为发送时间戳 redisTemplate.opsForZSet().add(convKey, String.valueOf(message.getMessageId()), message.getCreateTime().getTime()); // 裁剪ZSET，只保留最近500条消息ID，防止key无限膨胀 redisTemplate.opsForZSet().removeRange(convKey, 0, -501); }

ZSET 最妙的地方在于天然支持"按时间范围取数据"，配合游标分页非常丝滑。用户下拉加载更多时，用ZREVRANGEBYSCORE按时间倒序取消息 ID 列表，再批量查消息内容。

public List<ImMessageDTO> getHistoryMessages(String conversationId, Long cursor, int limit) { String convKey = "im:conv:" + conversationId; // 按时间倒序，取小于cursor时间戳的limit条消息ID Set<String> msgIds = redisTemplate.opsForZSet() .reverseRangeByScore(convKey, 0, cursor - 1, 0, limit); List<ImMessageDTO> result = new ArrayList<>(); for (String msgId : msgIds) { String msgKey = "im:msg:" + msgId; String msgJson = redisTemplate.opsForValue().get(msgKey); if (msgJson != null) { result.add(JSON.parseObject(msgJson, ImMessageDTO.class)); } } // 如果缓存未命中部分消息，回源MySQL补齐 fillFromDatabase(result, msgIds, conversationId); return result; }

4.2 缓存穿透与击穿的防护手段

私信场景下缓存穿透的典型场景是：用户打开一个很旧的会话，缓存里只有最近 500 条消息，用户继续往上翻，ZSET 里已经取不到更早的消息 ID。如果每次翻页都直接查 MySQL，热点会话的深分页查询会拖慢数据库。

我的处理方案是"缓存回源标记"：当 ZSET 取回来的消息 ID 数量小于请求的 limit 时，说明已经到了缓存边界，此时直接走 MySQL 查询完整历史。为了避免同一个会话被并发穿透查询打爆数据库，加了一个简单的互斥锁。

public List<ImMessageDTO> getHistoryFromDatabase(String conversationId, Long cursor, int limit) { // 加锁防止缓存击穿 String lockKey = "im:lock:conv:" + conversationId; boolean locked = redisTemplate.opsForValue() .setIfAbsent(lockKey, "1", 3, TimeUnit.SECONDS); if (!locked) { // 拿不到锁说明有其他线程正在回源，短暂等待后重试 Thread.sleep(50); return getHistoryMessages(conversationId, cursor, limit); } try { // 查MySQL历史消息 // 顺便把查到的消息重新写回缓存 } finally { redisTemplate.delete(lockKey); } }

还有一个隐蔽的坑：Redis 的removeRange裁剪 ZSET 时，如果消息发送时间戳相同（同一台机器同一毫秒），ZSET 的 score 会相同，Redis 内部会按 member（消息 ID）字典序排序。这种情况下裁剪可能误删同 score 的消息。解决方法是 score 用时间戳 * 1000 + 消息ID后缀，确保同毫秒内的消息也有不同的 score。

4.3 缓存与 MySQL 的一致性保障

聊到缓存就绕不开一致性问题。我的策略是"Cache Aside + 延迟双删"的简化版本。因为私信消息是"只追加、不修改"的数据（撤回是极低频操作），所以一致性天然比普通业务系统好做：

写入侧：先在 MySQL 落库，再通过 MQ 消费者写 Redis。因为 MQ 是 FIFO 的，同一会话的消息写 Redis 的顺序和写 MySQL 的顺序保持一致，不会出现缓存乱序。
更新侧（撤回场景）：先更新 MySQL 状态为撤回，再删除 Redis 中的消息内容 Key。这样下次读取时缓存未命中，回源 MySQL 拿到的是已撤回状态，然后重建缓存。

这里需要注意消息的"发送中"状态处理。接口层先写 MySQL 状态为SENDING，然后投递 MQ，消费者再把状态改为SENT。如果这时候用户立刻刷新聊天记录，可能读到状态为SENDING的消息。我的处理是在查询接口做一个小优化：如果消息状态是SENDING且创建时间超过 30 秒，说明 MQ 消费可能失败了，查询接口会直接返回"消息发送失败"的提示，同时触发一次补偿投递。

5. 实战复盘：压测结果与性能瓶颈定位

这套系统上线前我做了两轮压测，第一轮暴露了不少问题，这里把最有价值的几个问题及调优过程写出来。

5.1 消息积压导致延迟飙升

第一轮压测时，我用 200 个并发线程持续发送私信，发现消息从发送到接收的延迟从最初的 200ms 一路飙升到 5 秒以上。排查后定位到两个瓶颈：

第一个是 RabbitMQ 消费者的prefetch设置不合理。默认情况下 RabbitMQ 会给消费者一次推送无限多条消息，消费者处理不过来，消息全堆积在本地内存里。我把prefetch调整为 50，意思是最多同时处理 50 条未确认消息，让消费者按拉模式逐个处理。这个改动立竿见影，延迟降到了 800ms 左右。

spring: rabbitmq: listener: simple: prefetch: 50 concurrency: 10 max-concurrency: 30 acknowledge-mode: manual

第二个瓶颈在 Redis 的批量写入。消费者每消费一条消息就同步写一次 Redis，高峰期每秒要执行上千次SET和ZADD。后来我用pipeline把批量消息的 Redis 写入合并成一次网络往返，性能提升非常明显。

public void batchCacheMessages(List<ImMessageDTO> messages) { redisTemplate.executePipelined((RedisCallback<Object>) connection -> { for (ImMessageDTO msg : messages) { String convKey = "im:conv:" + msg.getConversationId(); String msgKey = "im:msg:" + msg.getMessageId(); connection.stringCommands().set( msgKey.getBytes(), JSON.toJSONString(msg).getBytes()); connection.zSetCommands().zAdd( convKey.getBytes(), msg.getCreateTime().getTime(), String.valueOf(msg.getMessageId()).getBytes()); } return null; }); }

5.2 数据库连接池被打满

压测中另一个严重问题是 MySQL 连接池被打满。排查后发现罪魁祸首是"回源查询"太频繁。缓存边界判断的条件不够严格，导致大量请求穿透到了 MySQL。

优化方案有两层。第一层是扩大缓存容量，把 ZSET 的removeRange裁剪数量从 500 调整到 2000，让更多历史消息留在缓存里。第二层是加了一层"轻量索引缓存"：在 Redis 中额外存一个im:conv:meta:{conversation_id}字符串，记录该会话在缓存中的最早一条消息 ID。查询时如果游标大于这个最早 ID，说明缓存覆盖范围内，直接走 Redis；只有游标小于最早 ID 时才回源 MySQL。

public List<ImMessageDTO> getHistoryMessages(String conversationId, Long cursor, int limit) { String metaKey = "im:conv:meta:" + conversationId; String earliestMsgId = redisTemplate.opsForValue().get(metaKey); if (earliestMsgId != null && Long.parseLong(earliestMsgId) <= cursor) { // 游标在缓存覆盖范围内，走Redis return getHistoryFromCache(conversationId, cursor, limit); } // 超出缓存范围，回源MySQL return getHistoryFromDatabase(conversationId, cursor, limit); }

5.3 已读状态延迟过高

已读状态的体验指标是"发送方要尽快看到已读回执"。我最初的设计是已读上报只写 Redis，定时任务每 3 秒批量落库。但发送方侧边栏的"已读"标记依赖 MySQL 数据，导致用户看到已读回执最迟要等 3 秒。

优化方案是：发送方查询会话列表时，已读状态直接从 Redis 读取，而不是查 MySQL。因为 Redis 中im:read:{conversation_id}:{user_id}存的已读位置是实时的，MySQL 落库只是为了保证数据持久性。这样已读回执的展示延迟从 3 秒降到了 100ms 以内，实时性一下子就上来了。

6. 上线后的稳定性保障与监控体系

系统上线只是开始，真正考验人的是线上稳定性。这里分享三个我在后续运维中持续优化的方向。

6.1 消息补偿机制

任何消息队列都不敢保证 100% 不丢消息。我上线后遇到过一次 RabbitMQ 节点重启，少量消息在持久化之前丢失。虽然概率极低，但私信场景对消息完整性要求很高，必须要有补偿机制。

我的做法是：接口层落库 MySQL 时，在im_message表里存一个mq_status字段（0 未投递、1 已投递、2 已消费）。同时启动一个定时任务，每 5 分钟扫描一次mq_status = 0且创建时间超过 5 分钟的消息，重新投递到 RabbitMQ。

@Component public class MessageCompensationTask { @Scheduled(fixedDelay = 300000) public void compensate() { // 查询 mq_status = 0 且 create_time < now() - 5分钟 的消息 List<ImMessage> pendingMessages = messageMapper.selectPendingMessages(); for (ImMessage msg : pendingMessages) { try { rabbitTemplate.convertAndSend( RabbitMQConfig.IM_EXCHANGE, RabbitMQConfig.IM_MESSAGE_ROUTING_KEY, msg); messageMapper.updateMqStatus(msg.getId(), 1); } catch (Exception e) { log.error("补偿投递失败: messageId={}", msg.getId(), e); } } } }

这里有个细节：补偿投递必须保证幂等。因为消息可能已经被消费者处理过，只是 MQ 状态没来得及更新。所以消费者在处理消息时，要先判断 MySQL 里该消息的状态是否为SENDING或SENT，如果已经是SENT，直接 ACK 不重复处理。

6.2 Redis 内存治理与淘汰策略

Redis 存储了消息内容、会话索引、已读状态、未读计数四类数据，内存增长很快。尤其是消息内容 Key，设置了 5 天过期，但如果某个大 V 用户的消息量巨大，Redis 内存还是会告急。

我的治理方案是分级缓存：热门会话的消息存 Redis（5 天），普通会话的消息存 Redis（1 天），冷门会话直接用 MySQL。判断热门与否的方式很简单，维护一个会话的最近活跃时间，活跃会话的缓存 TTL 自动续期，不活跃会话的缓存到期自动淘汰。

另外给 Redis 配置了allkeys-lru淘汰策略，内存不足时优先淘汰最久未使用的 Key。虽然这不是最稳妥的方案，但作为兜底保障是够用的。

6.3 核心监控指标与告警

最后说说监控。私信系统我重点盯四个指标：

RabbitMQ 队列积压数：im.message.queue深度超过 5000 就要告警，说明消费能力跟不上生产速度。
消息端到端延迟：从发送接口调用到消费者完成推送，用 Redis 记录每个消息的处理时间，超过 5 秒告警。
Redis 内存使用率：超过 70% 就要排查是否有 Key 异常增长。
MySQL 慢查询：重点监控im_read_record表的批量 UPDATE 和im_message表的深分页查询。

这些指标我全部通过定时任务上报，配合一套简单的告警规则。实测下来，对线上问题定位帮助最大的是"消息端到端延迟"这个指标，它能直接反映整条链路的健康状况，任何一环出问题都会在这个指标上体现出来。

6.4 一套可以直接抄作业的部署检查清单

根据我这次实战的经验，整理一个私信系统上线前的检查清单，每一条都是真实踩过坑换来的：

RabbitMQ 的队列、交换机、绑定关系是否都声明为 durable（持久化）？
消费者是否启用了手动 ACK？失败消息是否配置了死信队列转移？
消息表是否建了conversation_id + create_time联合索引？是否用了游标分页而不是 OFFSET 分页？
Redis 的 ZSET 缓存是否设置了裁剪上限？score 是否避免同毫秒冲突？
已读上报是否走了 Redis 去重？批量落库任务是否幂等？
补偿投递任务是否配置？消费者是否做了重复消息判断？
发送接口是否用client_msg_id做了幂等？

这套清单我后来也用在团队其他项目的架构评审里，凡是涉及消息队列和缓存的系统，照着过一遍基本能堵住 90% 的坑。

私信系统做完后，我最大的体会是：这类业务看着简单，难点全在细节里。消息不丢靠的是落库顺序和补偿机制，已读实时靠的是 Redis 缓冲和异步落库，历史秒开靠的是 ZSET 缓存和分级淘汰。每一项单独拿出来都不难，难点在于把整个链路串起来时，如何保证一致性、性能和可靠性的平衡。如果这篇文章能帮你少踩几个坑，那就值了。最后再分享一个建议：任何涉及消息队列的系统，一定提前设计好死信队列和消息轨迹日志，线上出问题时能少熬夜。