引子
在构建实时聊天服务时，我们既要保证消息的即时传递，又需要对消息进行持久化存储以便查询历史记录。然而，直接同步写入数据库在高并发场景下容易成为性能瓶颈，影响消息的实时性。秉承"没有什么问题是加一层解决不了的"理念，引入消息队列（MQ）进行异步存储是一个优雅的解决方案。消息先快速写入MQ确保即时送达，随后由专门的消费者服务从队列取出，平稳写入数据库。
在本文中，我们将详细探讨如何利用 Spring Boot 3 结合消息队列技术，构建一个高效可靠的聊天消息存储系统。
关于MQ
MQ在这里主要的作用是实现解耦，将聊天功能与聊天内容的存储过程分离。这种机制很像工厂与批发商之间的订货关系优化——传统模式下，工厂每次出货都需要逐一通知各个批发商。
而引入MQ后，这一流程变得优雅高效，就像工厂只需在一个微信群里发布消息，所有批发商便能同时获取信息，无需一对一通知。工厂专注生产，批发商按需处理，两端各司其职。
消息队列作为服务间通信的中间媒介，在分布式系统中扮演着至关重要的角色。常见的解决方案有专业的消息队列系统（如RabbitMQ、Kafka、RocketMQ等）、分布式协调服务Zookeeper，以及基于Redis实现的轻量级队列。
MQ选型
在众多消息队列产品中，各有其特点和适用场景：
| 消息队列 | 开发语言 | 特点 | 适用场景 |
|---|---|---|---|
| RabbitMQ | Erlang | 成熟稳定、易于部署、丰富的路由功能、社区活跃 | 复杂路由需求、中小规模消息量、需要可靠性保证 |
| ActiveMQ | Java | 老牌MQ、JMS实现、资源消耗较高 | 传统企业应用、与Java生态紧密结合 |
| RocketMQ | Java | 高吞吐、低延迟、金融级可靠性、支持大量堆积 | 大规模互联网应用、金融支付场景 |
| Kafka | Scala/Java | 超高吞吐量、持久化、分区设计、擅长流处理 | 日志收集、大数据实时处理、流数据分析 |
| ZeroMQ | C++ | 轻量级、无中心化、嵌入式库 | 对性能极为敏感的场景、点对点通信 |
| Redis队列 | C | 轻量简单、基于内存、低延迟 | 简单场景、临时队列、对持久化要求不高 |
对于我们的聊天消息存储场景，最终选择了 RabbitMQ，主要基于以下考虑：
- 成熟稳定：RabbitMQ历史悠久，生产环境验证充分，可靠性有保障
- 灵活路由：提供丰富的交换机类型和绑定机制，可针对不同类型消息实现精细化路由
- 易于集成：与Spring生态深度整合，Spring Boot 提供了完善的 starter 支持
- 运维友好：部署简单，自带管理界面，便于监控和管理
- 社区支持：活跃的社区和丰富的文档资源，遇到问题容易找到解决方案
虽然在极高并发场景下 Kafka 或 RocketMQ 可能有更好的吞吐性能，但考虑到我们这里重点在系统的解耦上，RabbitMQ 已经能够很好地满足需求，同时降低了开发和维护成本。
应用场景
消息队列在系统架构中有多种经典应用场景：
异步处理：将耗时操作（如邮件发送、日志处理）交由消息队列异步处理，快速响应用户请求，提升体验。
性能提升：通过异步解耦，减少系统响应时间，提高吞吐量，尤其适合I/O密集型操作。
系统解耦：降低服务间直接依赖，提高系统弹性和可维护性，便于独立扩展和升级。
削峰填谷：在流量高峰期，消息队列可缓存请求，按处理能力逐步消费，防止系统过载崩溃。
在聊天消息存储场景中，我们主要利用RabbitMQ实现消息异步存储，既保证了聊天功能的响应速度，又能可靠地将消息持久化到数据库，同时为系统提供了应对消息高峰的能力。
关于RabbitMQ
一条消息在RabbitMQ中的完整生命周期如下：
- 生产者创建消息：在聊天应用中，用户发送一个聊天内容，应用将其封装成MQ消息
- 投递到交换机：生产者将消息发送到指定的Exchange，同时指定路由键(Routing Key)
- 交换机路由转发：Exchange根据消息的路由键和绑定规则，决定将消息投递到哪个队列
- 若是Direct交换机，则精确匹配路由键
- 若是Fanout交换机，则广播给所有绑定队列
- 若是Topic交换机，则按模式匹配路由
- 存入队列：符合条件的队列接收并存储消息，等待消费者处理
- 消费者获取消息：存储服务作为消费者从队列中获取消息，可以是推模式(Push)或拉模式(Pull)
- 处理确认：消费者成功处理消息后(如将聊天内容存入数据库)，向RabbitMQ发送确认(ACK)
- 消息删除：收到确认后，RabbitMQ从队列中删除该消息
安装RabbitMQ
RabbitMQ的安装可以通过多种方式进行，而Docker提供了最便捷的部署方案。以下是使用Docker快速部署RabbitMQ的步骤：
1. 拉取镜像
首先从Docker Hub拉取RabbitMQ官方镜像，建议选择带management标签的版本，它包含了Web管理界面，便于后续的可视化操作和监控：
docker pull rabbitmq:4.1-management
提示：各位读者在实操时可以访问Docker Hub查看并使用最新的版本
2. 启动容器
拉取镜像后，通过以下命令启动RabbitMQ容器：
docker run --name rabbitmq -p 5681:5671 -p 5682:5672 -p 4379:4369 -p 15681:15671 -p 15682:15672 -p 25682:25672 --restart always -d rabbitmq:4.1-management
这里我们做了以下映射和配置：
- 暴露AMQP端口(5672)和管理界面端口(15672)
- 配置容器自动重启(--restart always)，确保服务器重启后RabbitMQ也能自动启动
- 后台运行容器(-d)
3. 验证安装
启动成功后，在浏览器中访问http://127.0.0.1:15682打开RabbitMQ管理控制台：
使用默认的用户名和密码登录（均为guest）：
登录成功后，您将看到RabbitMQ的管理界面，可以在这里创建交换机、队列、查看连接状态以及监控消息吞吐量等重要指标。
注意：默认的guest用户只能从localhost访问，如需远程访问，建议创建新的管理员用户并设置适当的权限。
Spring Boot 整合 RabbitMQ
在开始之前，我们先创建消息表。本文的聊天服务基于之前的文章《Java 工程师进阶必备：Spring Boot 3 + Netty 构建高并发即时通讯服务》，感兴趣的读者可以自行查阅。
DROP TABLE IF EXISTS `chat_message`;
CREATE TABLE `chat_message` (
`id` varchar(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
`sender_id` varchar(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '发送者的用户id',
`receiver_id` varchar(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '接受者的用户id',
`receiver_type` int(11) NULL DEFAULT NULL COMMENT '消息接受者的类型，可以作为扩展字段',
`msg` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '聊天内容',
`msg_type` int(11) NOT NULL COMMENT '消息类型，有文字类、图片类、视频类...等，详见枚举类',
`chat_time` datetime NOT NULL COMMENT '消息的聊天时间，既是发送者的发送时间、又是接受者的接受时间',
`show_msg_date_time_flag` int(11) NULL DEFAULT NULL COMMENT '标记存储数据库，用于历史展示。每超过1分钟，则显示聊天时间，前端可以控制时间长短(扩展字段)',
`video_path` varchar(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL DEFAULT NULL COMMENT '视频地址',
`video_width` int(11) NULL DEFAULT NULL COMMENT '视频宽度',
`video_height` int(11) NULL DEFAULT NULL COMMENT '视频高度',
`video_times` int(11) NULL DEFAULT NULL COMMENT '视频时间',
`voice_path` varchar(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL DEFAULT NULL COMMENT '语音地址',
`speak_voice_duration` int(11) NULL DEFAULT NULL COMMENT '语音时长',
`is_read` tinyint(1) NULL DEFAULT NULL COMMENT '语音消息标记是否已读未读，true: 已读，false: 未读',
PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8mb4 COLLATE = utf8mb4_unicode_ci COMMENT = '聊天信息存储表' ROW_FORMAT = Dynamic;
导入依赖
首先，在项目的 pom.xml 文件中添加 RabbitMQ 依赖：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
添加配置
在 application.yml 或 application.properties 文件中添加 RabbitMQ 的配置：
spring:
rabbitmq:
host: 127.0.0.1
port: 5682
username: guest
password: guest
virtual-host: /
编写生产者
创建一个消息发布者类，用于发送消息到 RabbitMQ：
import com.pitayafruits.pojo.netty.ChatMsg;
import com.pitayafruits.utils.JsonUtils;
public class MessagePublisher {
// 定义交换机的名字
public static final String EXCHANGE = "pitayafruits_exchange";
// 定义队列的名字
public static final String QUEUE = "pitayafruits_queue";
// 发送信息到消息队列接受并且保存到数据库的路由地址
public static final String ROUTING_KEY_SEND = "pitayafruits.wechat.send";
public static void sendMsgToSave(ChatMsg msg) throws Exception {
RabbitMQConnectUtils connectUtils = new RabbitMQConnectUtils();
connectUtils.sendMsg(JsonUtils.objectToJson(msg),
EXCHANGE,
ROUTING_KEY_SEND);
}
}
编写发送消息的工具类
import com.rabbitmq.client.*;
import java.util.ArrayList;
import java.util.List;
public class RabbitMQConnectUtils {
private final List<Connection> connections = new ArrayList<>();
private final int maxConnection = 20;
// 开发环境 dev
private final String host = "127.0.0.1";
private final int port = 5682;
private final String username = "guest";
private final String password = "guest";
private final String virtualHost = "/";
public ConnectionFactory factory;
public ConnectionFactory getRabbitMqConnection() {
return getFactory();
}
public ConnectionFactory getFactory() {
initFactory();
return factory;
}
private void initFactory() {
try {
if (factory == null) {
factory = new ConnectionFactory();
factory.setHost(host);
factory.setPort(port);
factory.setUsername(username);
factory.setPassword(password);
factory.setVirtualHost(virtualHost);
}
} catch (Exception e) {
e.printStackTrace();
}
}
public void sendMsg(String message, String queue) throws Exception {
Connection connection = getConnection();
Channel channel = connection.createChannel();
channel.basicPublish("",
queue,
MessageProperties.PERSISTENT_TEXT_PLAIN,
message.getBytes("utf-8"));
channel.close();
setConnection(connection);
}
public void sendMsg(String message, String exchange, String routingKey) throws Exception {
Connection connection = getConnection();
Channel channel = connection.createChannel();
channel.basicPublish(exchange,
routingKey,
MessageProperties.PERSISTENT_TEXT_PLAIN,
message.getBytes("utf-8"));
channel.close();
setConnection(connection);
}
public GetResponse basicGet(String queue, boolean autoAck) throws Exception {
GetResponse getResponse = null;
Connection connection = getConnection();
Channel channel = connection.createChannel();
getResponse = channel.basicGet(queue, autoAck);
channel.close();
setConnection(connection);
return getResponse;
}
public Connection getConnection() throws Exception {
return getAndSetConnection(true, null);
}
public void setConnection(Connection connection) throws Exception {
getAndSetConnection(false, connection);
}
private synchronized Connection getAndSetConnection(boolean isGet, Connection connection) throws Exception {
getRabbitMqConnection();
if (isGet) {
if (connections.isEmpty()) {
return factory.newConnection();
}
Connection newConnection = connections.get(0);
connections.remove(0);
if (newConnection.isOpen()) {
return newConnection;
} else {
return factory.newConnection();
}
} else {
if (connections.size() < maxConnection) {
connections.add(connection);
}
return null;
}
}
}
编写消费者
创建一个消息消费者类，用于接收并处理消息：
import com.pitayafruits.pojo.netty.ChatMsg;
import com.pitayafruits.service.ChatMessageService;
import com.pitayafruits.utils.JsonUtils;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
/**
* @Auther 风间影月
*/
@Component
@Slf4j
public class RabbitMQConsumer {
@Resource
private ChatMessageService chatMessageService;
@RabbitListener(queues = {
RabbitMQConfig.QUEUE})
public void watchQueue(String payload, Message message) {
String routingKey = message.getMessageProperties().getReceivedRoutingKey();
log.info("routingKey = " + routingKey);
if (routingKey.equals(RabbitMQConfig.ROUTING_KEY_SEND)) {
String msg = payload;
ChatMsg chatMsg = JsonUtils.jsonToPojo(msg, ChatMsg.class);
chatMessageService.saveMsg(chatMsg);
}
}
方法调用
完成上述封装后，在本次的案例中，直接在聊天服务的发送消息方法中调用消息发布功能即可。
// 把聊天信息作为mq的消息发送给消费者进行消费处理(保存到数据库)
MessagePublisher.sendMsgToSave(chatMsg);
小结
通过 Spring Boot 整合 RabbitMQ，我们实现了消息的异步处理机制，将聊天消息的存储操作解耦，提高了系统的性能和可扩展性。当用户发送消息时，我们将消息发送到 RabbitMQ，然后由消费者异步处理并保存到数据库中，避免了直接操作数据库导致的性能瓶颈。
