如何设计一个站内消息系统？

// 未读计数操作
public long getUnreadCount(Long userId, int msgType) {
    String key = String.format("unread:%d:%d", userId, msgType);
    return redisTemplate.opsForValue().increment(key, 0);
}

public void markAsRead(Long userId, Long messageId, int msgType) {
    // 1. Redis 原子减 1
    String key = String.format("unread:%d:%d", userId, msgType);
    Long count = redisTemplate.opsForValue().decrement(key);
    if (count != null && count < 0) {
        redisTemplate.opsForValue().set(key, 0L);
    }
    // 2. 异步更新 DB（MySQL 批量标记已读）
    mqProducer.send(new MarkReadEvent(userId, messageId));
}

@Service
public class MessageService {

    @Autowired
    private MessageMapper messageMapper;
    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    @Autowired
    private RocketMQTemplate mqTemplate;

    @Transactional
    public SendMessageResult send(SendMessageRequest request) {
        // 1. 写入消息主表
        Message message = new Message();
        message.setMsgType(request.getMsgType());
        message.setTitle(request.getTitle());
        message.setContent(request.getContent());
        messageMapper.insert(message);

        // 2. 批量写入收件关系（群发场景分批插入）
        List<Long> recipientIds = request.getRecipientIds();
        batchInsertRecipients(message.getId(), recipientIds);

        // 3. 写入 Redis 收件箱 + 未读计数（批量 Pipeline）
        redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            for (Long recipientId : recipientIds) {
                String inboxKey = String.format("inbox:%d:%d", recipientId, request.getMsgType());
                connection.lPush(inboxKey.getBytes(),
                    String.valueOf(message.getId()).getBytes());
                // 限制收件箱长度，只保留最近 100 条
                connection.lTrim(inboxKey.getBytes(), 0, 99);

                String unreadKey = String.format("unread:%d:%d", recipientId, request.getMsgType());
                connection.incr(unreadKey.getBytes());

                String totalKey = String.format("unread:%d:total", recipientId);
                connection.incr(totalKey.getBytes());
            }
            return null;
        });

        // 4. 异步通知推送服务
        mqTemplate.convertAndSend("message-push-topic",
            new PushEvent(message.getId(), recipientIds));

        return new SendMessageResult(message.getId());
    }
}

public PageResult<MessageVO> listMessages(Long userId, int msgType, int page, int size) {
    String inboxKey = String.format("inbox:%d:%d", userId, msgType);

    // 1. 从 Redis 收件箱获取 messageId 列表
    List<String> msgIds = redisTemplate.opsForList().range(
        inboxKey, (long) (page - 1) * size, (long) page * size - 1);

    if (msgIds != null && !msgIds.isEmpty()) {
        // 2. 批量获取消息详情（先 Redis 缓存，miss 再查 DB）
        List<MessageVO> messages = getMessageDetailBatch(msgIds);
        return new PageResult<>(messages, getTotalCount(userId, msgType));
    }

    // 3. Redis 无数据，走 DB 分页
    return messageMapper.selectByRecipientPage(userId, msgType, page, size);
}

// 订阅 Redis 频道，收到消息后推送给本节点维护的在线用户
@RocketMQMessageListener(topic = "message-push-topic", consumerGroup = "push-group")
public class PushConsumer implements RocketMQListener<PushEvent> {

    @Autowired
    private WebSocketSessionManager sessionManager;

    @Override
    public void onMessage(PushEvent event) {
        for (Long userId : event.getRecipientIds()) {
            WebSocketSession session = sessionManager.getSession(userId);
            if (session != null && session.isOpen()) {
                // 本节点有该用户的连接，直接推送
                session.sendMessage(new TextMessage(
                    JsonUtils.toJson(new PushNotification(event.getMessageId()))
                ));
            }
            // 其他节点有该用户的连接，由 Pub/Sub 广播处理
            // 或者直接查 Redis 中的在线节点标识
        }
    }
}

// 分片异步批量插入
@Async("messageExecutor")
public void batchInsertRecipients(Long messageId, List<Long> recipientIds) {
    Lists.partition(recipientIds, 1000).forEach(batch -> {
        messageMapper.batchInsertRecipient(messageId, batch);
    });
}

// 定时对账任务（每天凌晨低峰期执行）
@Scheduled(cron = "0 0 3 * * ?")
public void syncUnreadCount() {
    // 从 DB 查出真实未读数，修复 Redis
    List<UnreadStat> stats = messageMapper.selectUnreadStats();
    stats.forEach(stat -> {
        String key = String.format("unread:%d:%d", stat.getUserId(), stat.getMsgType());
        redisTemplate.opsForValue().set(key, (long) stat.getUnreadCount());
    });
}

@PostMapping("/batch-read")
public Result batchMarkAsRead(@RequestParam int msgType) {
    Long userId = getCurrentUserId();
    // 1. Redis 批量清零
    String unreadKey = String.format("unread:%d:%d", userId, msgType);
    redisTemplate.opsForValue().set(unreadKey, 0L);
    String totalKey = String.format("unread:%d:total", userId);
    redisTemplate.opsForValue().decrement(totalKey);

    // 2. 异步更新 DB
    mqTemplate.convertAndSend("message-read-topic",
        new BatchReadEvent(userId, msgType));
    return Result.success();
}