RabbitMQ 部署、配置及应用示例
本指南将介绍如何在 Linux 环境下使用 Docker 部署 RabbitMQ，并配置 Spring Boot 应用以实现消息队列的集成和实际应用。
Docker 部署 RabbitMQ
请确保您的 Linux 系统（以 CentOS 7.8 为例）已安装 Docker。执行以下命令以启动 RabbitMQ 容器，并启用管理界面：
docker run -d \
--name rabbitmq_manager \
-e RABBITMQ_DEFAULT_USER=myuser \
-e RABBITMQ_DEFAULT_PASS=mypassword \
-p 15672:15672 \
-p 5672:5672 \
rabbitmq:3.9.7-management
部署完成后，请在服务器的网络安全组中开放以下端口：
4369: Erlang 端口发现5672: AMQP 客户端通信端口15672: 管理界面 UI 端口25672: RabbitMQ 集群节点间通信端口
您可以通过访问 http://<服务器IP>:15672 来打开 RabbitMQ 管理界面。
Spring Boot 引入 RabbitMQ 依赖
在您的 Spring Boot 项目的 pom.xml 文件中添加 spring-boot-starter-amqp 依赖：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
Spring Boot 配置文件
在 application.properties 或 application.yml 文件中配置 RabbitMQ 连接信息：
spring:
rabbitmq:
host: localhost
port: 5672
username: myuser
password: mypassword
virtual-host: /
listener:
simple:
# 消息确认模式: auto (自动确认) 或 manual (手动确认)
acknowledge-mode: auto
retry:
# 开启消费者消息重试机制
enabled: true
# 最大重试次数
max-attempts: 3
# 重试间隔时间 (毫秒)
max-interval: 5000
RabbitMQ 配置类
以下示例定义了一个 Topic 类型的交换机，以及两个队列，并使用通配符路由键将它们绑定到交换机。这种配置适用于需要根据消息内容进行分发和处理的场景。
package com.example.mq;
import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitMQConfig {
// 定义交换机名称
public static final String MY_EXCHANGE = "my_app.exchange";
// 定义队列名称
public static final String QUEUE_ONE = "my_app.queue.one";
public static final String QUEUE_TWO = "my_app.queue.two";
// 定义路由键模式
public static final String ROUTING_KEY_PATTERN_ONE = "data.process.*";
public static final String ROUTING_KEY_PATTERN_TWO = "data.*.details";
/**
* 配置Topic类型的交换机
*/
@Bean
public Exchange myExchange() {
return new TopicExchange(MY_EXCHANGE, true, false);
}
/**
* 配置第一个队列
*/
@Bean
public Queue queueOne() {
return new Queue(QUEUE_ONE, true, false, false);
}
/**
* 配置第二个队列
*/
@Bean
public Queue queueTwo() {
return new Queue(QUEUE_TWO, true, false, false);
}
/**
* 将第一个队列绑定到交换机，使用通配符路由键
*/
@Bean
public Binding bindingQueueOne() {
return BindingBuilder.bind(queueOne())
.to(myExchange())
.with(ROUTING_KEY_PATTERN_ONE);
}
/**
* 将第二个队列绑定到交换机，使用通配符路由键
*/
@Bean
public Binding bindingQueueTwo() {
return BindingBuilder.bind(queueTwo())
.to(myExchange())
.with(ROUTING_KEY_PATTERN_TWO);
}
}
消息消费者
以下是两个独立的消费者类，分别监听不同的队列。当消息到达各自的队列时，对应的消费者将被触发。
package com.example.mq.consumer;
import com.example.dto.EventMessage; // 假设这是您的消息体对象
import com.example.exception.BizException; // 假设这是您的自定义异常
import com.example.exception.BizCodeEnum; // 假设这是您的业务错误码
import com.example.service.MyDataProcessingService; // 假设这是您的业务服务类
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.io.IOException;
@Component
@Slf4j
@RabbitListener(queues = RabbitMQConfig.QUEUE_ONE) // 监听第一个队列
public class MyDataProcessingListener {
@Autowired
private MyDataProcessingService dataProcessingService;
@RabbitHandler
public void processMessage(EventMessage eventMessage, Message message, com.rabbitmq.client.Channel channel) throws IOException {
log.info("Received message for QUEUE_ONE: {}", eventMessage);
try {
// 假设 EventMessage 中有一个类型字段，用于区分具体处理逻辑
eventMessage.setEventType("PROCESS_DATA");
dataProcessingService.handleEvent(eventMessage);
log.info("Successfully processed message from QUEUE_ONE: {}", eventMessage);
// 手动确认消息（如果 acknowledge-mode 为 manual）
// channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
log.error("Failed to process message from QUEUE_ONE: {}, error: {}", eventMessage, e.getMessage());
// 业务异常处理
if (!(e instanceof BizException)) {
throw new BizException(BizCodeEnum.MQ_CONSUME_ERROR);
} else {
throw e; // 重新抛出业务异常
}
// 发生异常时，根据需要进行死信队列处理或重试
// channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false); // 拒绝消息，不重新入队
}
}
}
package com.example.mq.consumer;
import com.example.dto.EventMessage; // 假设这是您的消息体对象
import com.example.exception.BizException; // 假设这是您的自定义异常
import com.example.exception.BizCodeEnum; // 假设这是您的业务错误码
import com.example.service.MyDetailsProcessingService; // 假设这是另一个业务服务类
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.io.IOException;
@Component
@Slf4j
@RabbitListener(queues = RabbitMQConfig.QUEUE_TWO) // 监听第二个队列
public class MyDetailsProcessingListener {
@Autowired
private MyDetailsProcessingService detailsProcessingService;
@RabbitHandler
public void processDetails(EventMessage eventMessage, Message message, com.rabbitmq.client.Channel channel) throws IOException {
log.info("Received message for QUEUE_TWO: {}", eventMessage);
try {
eventMessage.setEventType("PROCESS_DETAILS");
detailsProcessingService.handleEventDetails(eventMessage);
log.info("Successfully processed message from QUEUE_TWO: {}", eventMessage);
// 手动确认消息 (如果 acknowledge-mode 为 manual)
// channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
log.error("Failed to process message from QUEUE_TWO: {}, error: {}", eventMessage, e.getMessage());
// 业务异常处理
if (!(e instanceof BizException)) {
throw new BizException(BizCodeEnum.MQ_CONSUME_ERROR);
} else {
throw e; // 重新抛出业务异常
}
// 发生异常时，根据需要进行死信队列处理或重试
// channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
}
}
}
消息生产者示例
在 Controller 或 Service 层，您可以使用 RabbitTemplate 来发送消息。
package com.example.controller;
import com.example.dto.EventMessage; // 您的消息体对象
import com.example.mq.RabbitMQConfig; // 引入配置类
import com.example.util.JsonData; // 假设这是您的统一返回对象
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("/api/v1/events")
public class EventController {
@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private RabbitMQConfig rabbitMQConfig; // 注入配置类以获取交换机和路由键
/**
* 发送一个事件消息
* @param requestData 包含事件信息的请求体
* @return 统一返回对象
*/
@PostMapping("send")
public JsonData sendEvent(@RequestBody EventMessage requestData) {
// 构建要发送的消息体
EventMessage messageToSend = EventMessage.builder()
// ... 设置 requestData 中的字段到 messageToSend ...
.build();
// 选择路由键，这里演示发送一个会匹配到 QUEUE_ONE 的路由键
String routingKey = "data.process.raw"; // 示例路由键
// 发送消息到 RabbitMQ
rabbitTemplate.convertAndSend(
RabbitMQConfig.MY_EXCHANGE, // 交换机名称
routingKey, // 路由键
messageToSend // 消息体
);
// 返回成功响应
return JsonData.buildSuccess("Event message sent successfully.");
}
/**
* 发送另一个事件消息，可能匹配到 QUEUE_TWO
* @param requestData 包含事件信息的请求体
* @return 统一返回对象
*/
@PostMapping("send-details")
public JsonData sendDetailsEvent(@RequestBody EventMessage requestData) {
EventMessage messageToSend = EventMessage.builder()
// ... 设置 requestData 中的字段到 messageToSend ...
.build();
// 选择一个可能匹配到 QUEUE_TWO 的路由键
String routingKey = "data.specific.details"; // 示例路由键
rabbitTemplate.convertAndSend(RabbitMQConfig.MY_EXCHANGE, routingKey, messageToSend);
return JsonData.buildSuccess("Details event message sent successfully.");
}
}
