Java中的RabbitMQ完全指南
1. 引言
什么是RabbitMQ
RabbitMQ是一个开源的消息代理和队列服务器，实现了高级消息队列协议（AMQP）。它充当应用程序之间的消息中间件，允许分布式系统中的不同组件进行异步通信。RabbitMQ使用Erlang语言开发，以其高性能、可靠性和可扩展性而闻名。
消息队列的核心概念
消息队列是一种异步通信机制，它允许应用程序通过发送和接收消息来进行通信，而不需要直接连接。这种模式带来了以下优势：
- 解耦：生产者和消费者不需要同时在线
- 可扩展性：可以独立扩展生产者和消费者
- 可靠性：消息可以持久化存储，确保不丢失
- 灵活性：支持多种消息传递模式
RabbitMQ的优势和应用场景
RabbitMQ在企业级应用中具有以下优势：
- 多协议支持：支持AMQP、STOMP、MQTT等多种协议
- 灵活的路由：支持多种Exchange类型和复杂的路由规则
- 集群支持：可以构建高可用的集群架构
- 管理界面：提供Web管理控制台
- 丰富的客户端库：支持多种编程语言
常见应用场景包括：微服务解耦、异步任务处理、系统集成、削峰填谷等。
2. RabbitMQ基础概念
Exchange（交换器）类型详解
Exchange是RabbitMQ的核心组件，负责接收生产者发送的消息并将其路由到相应的队列。主要有四种类型：
Direct Exchange（直连交换器）
- 根据routing key精确匹配路由消息
- 适用于单播消息传递
- 默认的交换器类型
Fanout Exchange（扇形交换器）
- 将消息广播到所有绑定的队列
- 忽略routing key
- 适用于广播场景
Topic Exchange（主题交换器）
- 基于通配符模式匹配routing key
- 支持"*“（单个单词）和”#"（零个或多个单词）
- 灵活的路由规则
Headers Exchange（头交换器）
- 基于消息头属性进行路由
- 较少使用，性能相对较低
Queue（队列）和消息持久化
Queue是存储消息的容器，具有以下特性：
- FIFO原则：先进先出的消息处理顺序
- 持久化：可以配置队列和消息的持久化
- 排他性：可以设置队列只能被一个连接使用
- 自动删除：当没有消费者时自动删除队列
Routing Key和Binding
- Routing Key：生产者发送消息时指定的路由键
- Binding：Exchange和Queue之间的绑定关系
- Binding Key：绑定时指定的键，用于匹配routing key
Virtual Host（虚拟主机）
Virtual Host提供了逻辑隔离，类似于网络中的虚拟主机概念：
- 不同vhost中的Exchange、Queue等资源完全隔离
- 每个vhost有独立的权限控制
- 默认vhost为"/"
3. Java环境准备
RabbitMQ服务器安装配置
Docker方式安装（推荐）：
# 拉取RabbitMQ镜像（包含管理插件）
docker pull rabbitmq:3-management
# 运行RabbitMQ容器
docker run -d --name rabbitmq \
-p 5672:5672 \
-p 15672:15672 \
-e RABBITMQ_DEFAULT_USER=admin \
-e RABBITMQ_DEFAULT_PASS=admin123 \
rabbitmq:3-management
安装完成后，可以通过 http://localhost:15672 访问管理界面。
Maven/Gradle依赖配置
Maven配置：
<dependencies>
<!-- RabbitMQ Java客户端 -->
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.18.0</version>
</dependency>
<!-- Spring Boot RabbitMQ Starter -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!-- 日志依赖 -->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-simple</artifactId>
<version>2.0.7</version>
</dependency>
</dependencies>
Gradle配置：
dependencies {
implementation 'com.rabbitmq:amqp-client:5.18.0'
implementation 'org.springframework.boot:spring-boot-starter-amqp'
implementation 'org.slf4j:slf4j-simple:2.0.7'
}
连接工厂和连接管理
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class RabbitMQConnection {
public static Connection getConnection() throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setUsername("admin");
factory.setPassword("admin123");
factory.setVirtualHost("/");
// 连接超时设置
factory.setConnectionTimeout(30000);
factory.setRequestedHeartbeat(60);
return factory.newConnection();
}
}
4. 基础消息模式实现
Simple Queue（简单队列）
最基本的消息模式，一个生产者发送消息到队列，一个消费者从队列接收消息。
生产者代码：
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
public class SimpleProducer {
private final static String QUEUE_NAME = "simple_queue";
public static void main(String[] args) throws Exception {
try (Connection connection = RabbitMQConnection.getConnection();
Channel channel = connection.createChannel()) {
// 声明队列
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
// 发送消息
String message = "Hello, RabbitMQ!";
channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));
System.out.println("Sent: " + message);
}
}
}
消费者代码：
import com.rabbitmq.client.*;
public class SimpleConsumer {
private final static String QUEUE_NAME = "simple_queue";
public static void main(String[] args) throws Exception {
Connection connection = RabbitMQConnection.getConnection();
Channel channel = connection.createChannel();
// 声明队列
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
// 创建消费者
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
System.out.println("Received: " + message);
};
// 开始消费
channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {});
System.out.println("Waiting for messages...");
}
}
Work Queues（工作队列）
工作队列模式用于在多个消费者之间分发耗时任务，实现负载均衡。
生产者代码：
public class WorkQueueProducer {
private final static String QUEUE_NAME = "work_queue";
public static void main(String[] args) throws Exception {
try (Connection connection = RabbitMQConnection.getConnection();
Channel channel = connection.createChannel()) {
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
// 发送多个任务
for (int i = 1; i <= 10; i++) {
String message = "Task " + i;
channel.basicPublish("", QUEUE_NAME,
MessageProperties.PERSISTENT_TEXT_PLAIN,
message.getBytes("UTF-8"));
System.out.println("Sent: " + message);
}
}
}
}
消费者代码：
public class WorkQueueConsumer {
private final static String QUEUE_NAME = "work_queue";
public static void main(String[] args) throws Exception {
Connection connection = RabbitMQConnection.getConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
// 设置每次只处理一个消息（公平分发）
channel.basicQos(1);
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
System.out.println("Processing: " + message);
try {
// 模拟耗时处理
Thread.sleep(2000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
// 手动确认消息
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
System.out.println("Completed: " + message);
};
// 关闭自动确认
channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});
System.out.println("Worker waiting for tasks...");
}
}
Publish/Subscribe（发布订阅）
发布订阅模式使用fanout类型的Exchange将消息广播到所有绑定的队列。
发布者代码：
public class PublishSubscribeProducer {
private final static String EXCHANGE_NAME = "fanout_exchange";
public static void main(String[] args) throws Exception {
try (Connection connection = RabbitMQConnection.getConnection();
Channel channel = connection.createChannel()) {
// 声明fanout类型的交换器
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
String message = "Broadcast message to all subscribers!";
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
System.out.println("Published: " + message);
}
}
}
订阅者代码：
public class PublishSubscribeConsumer {
private final static String EXCHANGE_NAME = "fanout_exchange";
public static void main(String[] args) throws Exception {
Connection connection = RabbitMQConnection.getConnection();
Channel channel = connection.createChannel();
// 声明交换器
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
// 创建临时队列
String queueName = channel.queueDeclare().getQueue();
// 将队列绑定到交换器
channel.queueBind(queueName, EXCHANGE_NAME, "");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
System.out.println("Received: " + message);
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});
System.out.println("Subscriber waiting for messages...");
}
}
Routing（路由模式）
路由模式使用direct类型的Exchange根据routing key将消息路由到特定队列。
生产者代码：
public class RoutingProducer {
private final static String EXCHANGE_NAME = "direct_exchange";
public static void main(String[] args) throws Exception {
try (Connection connection = RabbitMQConnection.getConnection();
Channel channel = connection.createChannel()) {
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
// 发送不同级别的日志消息
String[] levels = {"info", "warning", "error"};
for (String level : levels) {
String message = "This is a " + level + " message";
channel.basicPublish(EXCHANGE_NAME, level, null, message.getBytes("UTF-8"));
System.out.println("Sent [" + level + "]: " + message);
}
}
}
}
消费者代码：
public class RoutingConsumer {
private final static String EXCHANGE_NAME = "direct_exchange";
public static void main(String[] args) throws Exception {
Connection connection = RabbitMQConnection.getConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
String queueName = channel.queueDeclare().getQueue();
// 只接收error级别的消息
channel.queueBind(queueName, EXCHANGE_NAME, "error");
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
String routingKey = delivery.getEnvelope().getRoutingKey();
System.out.println("Received [" + routingKey + "]: " + message);
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});
System.out.println("Consumer waiting for error messages...");
}
}
Topics（主题模式）
主题模式使用topic类型的Exchange支持通配符路由。
生产者代码：
public class TopicProducer {
private final static String EXCHANGE_NAME = "topic_exchange";
public static void main(String[] args) throws Exception {
try (Connection connection = RabbitMQConnection.getConnection();
Channel channel = connection.createChannel()) {
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
// 发送不同主题的消息
String[] routingKeys = {
"user.info.create",
"user.warning.update",
"order.error.payment",
"system.info.startup"
};
for (String routingKey : routingKeys) {
String message = "Message for " + routingKey;
channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));
System.out.println("Sent [" + routingKey + "]: " + message);
}
}
}
}
消费者代码：
public class TopicConsumer {
private final static String EXCHANGE_NAME = "topic_exchange";
public static void main(String[] args) throws Exception {
Connection connection = RabbitMQConnection.getConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
String queueName = channel.queueDeclare().getQueue();
// 绑定多个模式
channel.queueBind(queueName, EXCHANGE_NAME, "user.*.*"); // 所有用户相关消息
channel.queueBind(queueName, EXCHANGE_NAME, "*.error.*"); // 所有错误消息
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
String routingKey = delivery.getEnvelope().getRoutingKey();
System.out.println("Received [" + routingKey + "]: " + message);
};
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});
System.out.println("Consumer waiting for messages matching patterns...");
}
}
RPC模式
RPC模式实现远程过程调用，客户端发送请求并等待服务端响应。
RPC服务端：
public class RPCServer {
private final static String QUEUE_NAME = "rpc_queue";
public static void main(String[] args) throws Exception {
Connection connection = RabbitMQConnection.getConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
channel.basicQos(1);
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
AMQP.BasicProperties replyProps = new AMQP.BasicProperties
.Builder()
.correlationId(delivery.getProperties().getCorrelationId())
.build();
String response = "";
try {
String message = new String(delivery.getBody(), "UTF-8");
int n = Integer.parseInt(message);
System.out.println("Computing fibonacci(" + n + ")");
response = String.valueOf(fibonacci(n));
} catch (RuntimeException e) {
System.out.println("Error: " + e.toString());
response = "Error: " + e.toString();
} finally {
channel.basicPublish("", delivery.getProperties().getReplyTo(),
replyProps, response.getBytes("UTF-8"));
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
};
channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});
System.out.println("RPC Server waiting for requests...");
}
private static int fibonacci(int n) {
if (n <= 1) return n;
return fibonacci(n - 1) + fibonacci(n - 2);
}
}
RPC客户端：
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
public class RPCClient implements AutoCloseable {
private Connection connection;
private Channel channel;
private String requestQueueName = "rpc_queue";
public RPCClient() throws Exception {
connection = RabbitMQConnection.getConnection();
channel = connection.createChannel();
}
public String call(String message) throws Exception {
final String corrId = UUID.randomUUID().toString();
String replyQueueName = channel.queueDeclare().getQueue();
AMQP.BasicProperties props = new AMQP.BasicProperties
.Builder()
.correlationId(corrId)
.replyTo(replyQueueName)
.build();
channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));
final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);
String ctag = channel.basicConsume(replyQueueName, true, (consumerTag, delivery) -> {
if (delivery.getProperties().getCorrelationId().equals(corrId)) {
response.offer(new String(delivery.getBody(), "UTF-8"));
}
}, consumerTag -> {});
String result = response.take();
channel.basicCancel(ctag);
return result;
}
@Override
public void close() throws Exception {
connection.close();
}
public static void main(String[] args) throws Exception {
try (RPCClient client = new RPCClient()) {
System.out.println("Requesting fibonacci(10)");
String response = client.call("10");
System.out.println("Got: " + response);
}
}
}
5. Spring Boot集成RabbitMQ
Spring AMQP配置
application.yml配置：
spring:
rabbitmq:
host: localhost
port: 5672
username: admin
password: admin123
virtual-host: /
connection-timeout: 30000
publisher-confirm-type: correlated
publisher-returns: true
listener:
simple:
acknowledge-mode: manual
concurrency: 2
max-concurrency: 10
配置类：
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 DIRECT_EXCHANGE = "spring.direct.exchange";
public static final String TOPIC_EXCHANGE = "spring.topic.exchange";
public static final String DIRECT_QUEUE = "spring.direct.queue";
public static final String TOPIC_QUEUE = "spring.topic.queue";
// 声明Direct Exchange
@Bean
public DirectExchange directExchange() {
return new DirectExchange(DIRECT_EXCHANGE, true, false);
}
// 声明Topic Exchange
@Bean
public TopicExchange topicExchange() {
return new TopicExchange(TOPIC_EXCHANGE, true, false);
}
// 声明队列
@Bean
public Queue directQueue() {
return QueueBuilder.durable(DIRECT_QUEUE).build();
}
@Bean
public Queue topicQueue() {
return QueueBuilder.durable(TOPIC_QUEUE).build();
}
// 绑定关系
@Bean
public Binding directBinding() {
return BindingBuilder.bind(directQueue())
.to(directExchange())
.with("direct.routing.key");
}
@Bean
public Binding topicBinding() {
return BindingBuilder.bind(topicQueue())
.to(topicExchange())
.with("topic.*.key");
}
}
RabbitTemplate使用
消息生产者服务：
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
@Service
public class MessageProducerService {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendDirectMessage(String message) {
rabbitTemplate.convertAndSend(
RabbitMQConfig.DIRECT_EXCHANGE,
"direct.routing.key",
message
);
System.out.println("Sent direct message: " + message);
}
public void sendTopicMessage(String routingKey, String message) {
rabbitTemplate.convertAndSend(
RabbitMQConfig.TOPIC_EXCHANGE,
routingKey,
message
);
System.out.println("Sent topic message with key " + routingKey + ": " + message);
}
// 发送对象消息
public void sendObjectMessage(Object obj) {
rabbitTemplate.convertAndSend(
RabbitMQConfig.DIRECT_EXCHANGE,
"direct.routing.key",
obj
);
}
}
@RabbitListener注解详解
消息消费者服务：
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;
import com.rabbitmq.client.Channel;
@Service
public class MessageConsumerService {
// 基础消费者
@RabbitListener(queues = RabbitMQConfig.DIRECT_QUEUE)
public void handleDirectMessage(String message) {
System.out.println("Received direct message: " + message);
}
// 手动确认消息
@RabbitListener(queues = RabbitMQConfig.TOPIC_QUEUE)
public void handleTopicMessage(
@Payload String message,
@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
Channel channel) throws Exception {
try {
System.out.println("Processing topic message: " + message);
// 模拟业务处理
Thread.sleep(1000);
// 手动确认
channel.basicAck(deliveryTag, false);
} catch (Exception e) {
System.err.println("Error processing message: " + e.getMessage());
// 拒绝消息并重新入队
channel.basicNack(deliveryTag, false, true);
}
}
// 接收完整消息对象
@RabbitListener(queues = RabbitMQConfig.DIRECT_QUEUE)
public void handleCompleteMessage(Message message, Channel channel) throws Exception {
String body = new String(message.getBody());
String routingKey = message.getMessageProperties().getReceivedRoutingKey();
System.out.println("Received message: " + body + " with routing key: " + routingKey);
// 手动确认
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}
}
消息转换器配置
JSON消息转换器：
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class MessageConverterConfig {
@Bean
public Jackson2JsonMessageConverter jsonMessageConverter() {
return new Jackson2JsonMessageConverter();
}
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate template = new RabbitTemplate(connectionFactory);
template.setMessageConverter(jsonMessageConverter());
// 配置确认回调
template.setConfirmCallback((correlationData, ack, cause) -> {
if (ack) {
System.out.println("Message sent successfully");
} else {
System.out.println("Message send failed: " + cause);
}
});
// 配置返回回调
template.setReturnsCallback(returned -> {
System.out.println("Message returned: " + returned.getMessage());
});
return template;
}
}
使用JSON转换器发送对象：
// 定义消息对象
public class UserMessage {
private Long id;
private String name;
private String email;
// 构造函数、getter、setter
public UserMessage() {}
public UserMessage(Long id, String name, String email) {
this.id = id;
this.name = name;
this.email = email;
}
// getter和setter方法...
}
// 发送和接收对象消息
@Service
public class UserMessageService {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendUserMessage(UserMessage user) {
rabbitTemplate.convertAndSend(
RabbitMQConfig.DIRECT_EXCHANGE,
"user.routing.key",
user
);
}
@RabbitListener(queues = "user.queue")
public void handleUserMessage(UserMessage user) {
System.out.println("Received user: " + user.getName() + " (" + user.getEmail() + ")");
}
}
6. 高级特性
消息确认机制（ACK）
RabbitMQ提供了多种消息确认机制来保证消息的可靠传递：
自动确认（Auto ACK）：
// 消息被消费者接收后立即确认
@RabbitListener(queues = "auto.ack.queue", ackMode = "AUTO")
public void handleAutoAck(String message) {
System.out.println("Auto ACK: " + message);
}
手动确认（Manual ACK）：
@RabbitListener(queues = "manual.ack.queue", ackMode = "MANUAL")
public void handleManualAck(
String message,
@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
Channel channel) throws Exception {
try {
// 处理业务逻辑
processMessage(message);
// 确认消息
channel.basicAck(deliveryTag, false);
} catch (Exception e) {
// 拒绝消息，重新入队
channel.basicNack(deliveryTag, false, true);
}
}
发布确认：
@Configuration
public class PublisherConfirmConfig {
@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate template = new RabbitTemplate(connectionFactory);
// 启用发布确认
template.setConfirmCallback((correlationData, ack, cause) -> {
if (ack) {
System.out.println("Message published successfully");
} else {
System.out.println("Message publish failed: " + cause);
}
});
// 启用消息返回
template.setReturnsCallback(returned -> {
System.out.println("Message returned: " + returned.getMessage().toString());
System.out.println("Reply code: " + returned.getReplyCode());
System.out.println("Reply text: " + returned.getReplyText());
});
return template;
}
}
死信队列（DLX）处理
死信队列用于处理无法正常消费的消息，常见的死信场景包括：
- 消息被拒绝且不重新入队
- 消息TTL过期
- 队列达到最大长度
死信队列配置：
@Configuration
public class DeadLetterConfig {
public static final String BUSINESS_EXCHANGE = "business.exchange";
public static final String BUSINESS_QUEUE = "business.queue";
public static final String DEAD_LETTER_EXCHANGE = "dlx.exchange";
public static final String DEAD_LETTER_QUEUE = "dlx.queue";
// 业务交换器
@Bean
public DirectExchange businessExchange() {
return new DirectExchange(BUSINESS_EXCHANGE);
}
// 死信交换器
@Bean
public DirectExchange deadLetterExchange() {
return new DirectExchange(DEAD_LETTER_EXCHANGE);
}
// 业务队列（配置死信交换器）
@Bean
public Queue businessQueue() {
return QueueBuilder
.durable(BUSINESS_QUEUE)
.withArgument("x-dead-letter-exchange", DEAD_LETTER_EXCHANGE)
.withArgument("x-dead-letter-routing-key", "dead.letter.routing.key")
.withArgument("x-message-ttl", 10000) // 消息TTL 10秒
.build();
}
// 死信队列
@Bean
public Queue deadLetterQueue() {
return QueueBuilder.durable(DEAD_LETTER_QUEUE).build();
}
// 绑定关系
@Bean
public Binding businessBinding() {
return BindingBuilder.bind(businessQueue())
.to(businessExchange())
.with("business.routing.key");
}
@Bean
public Binding deadLetterBinding() {
return BindingBuilder.bind(deadLetterQueue())
.to(deadLetterExchange())
.with("dead.letter.routing.key");
}
}
死信处理服务：
@Service
public class DeadLetterService {
@Autowired
private RabbitTemplate rabbitTemplate;
// 发送业务消息
public void sendBusinessMessage(String message) {
rabbitTemplate.convertAndSend(
DeadLetterConfig.BUSINESS_EXCHANGE,
"business.routing.key",
message
);
}
// 业务消息处理
@RabbitListener(queues = DeadLetterConfig.BUSINESS_QUEUE)
public void handleBusinessMessage(
String message,
@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
Channel channel) throws Exception {
try {
System.out.println("Processing business message: " + message);
// 模拟处理失败
if (message.contains("error")) {
throw new RuntimeException("Business processing failed");
}
channel.basicAck(deliveryTag, false);
} catch (Exception e) {
System.err.println("Business processing failed: " + e.getMessage());
// 拒绝消息，不重新入队，进入死信队列
channel.basicNack(deliveryTag, false, false);
}
}
// 死信消息处理
@RabbitListener(queues = DeadLetterConfig.DEAD_LETTER_QUEUE)
public void handleDeadLetterMessage(String message) {
System.out.println("Handling dead letter message: " + message);
// 记录日志、发送告警、人工处理等
logDeadLetterMessage(message);
}
private void logDeadLetterMessage(String message) {
// 实现日志记录逻辑
System.out.println("Dead letter logged: " + message);
}
}
消息TTL和队列过期
消息TTL配置：
@Service
public class TTLMessageService {
@Autowired
private RabbitTemplate rabbitTemplate;
// 发送带TTL的消息
public void sendTTLMessage(String message, int ttlSeconds) {
MessageProperties properties = new MessageProperties();
properties.setExpiration(String.valueOf(ttlSeconds * 1000)); // 毫秒
Message msg = new Message(message.getBytes(), properties);
rabbitTemplate.send("ttl.exchange", "ttl.routing.key", msg);
}
// 使用MessagePostProcessor设置TTL
public void sendTTLMessageWithProcessor(String message, int ttlSeconds) {
rabbitTemplate.convertAndSend(
"ttl.exchange",
"ttl.routing.key",
message,
msg -> {
msg.getMessageProperties().setExpiration(String.valueOf(ttlSeconds * 1000));
return msg;
}
);
}
}
队列TTL配置：
@Bean
public Queue ttlQueue() {
return QueueBuilder
.durable("ttl.queue")
.withArgument("x-message-ttl", 60000) // 队列中消息的默认TTL
.withArgument("x-expires", 300000) // 队列没有消费者时的过期时间
.build();
}
优先级队列
优先级队列配置：
@Bean
public Queue priorityQueue() {
return QueueBuilder
.durable("priority.queue")
.withArgument("x-max-priority", 10) // 最大优先级为10
.build();
}
发送优先级消息：
@Service
public class PriorityMessageService {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendPriorityMessage(String message, int priority) {
rabbitTemplate.convertAndSend(
"priority.exchange",
"priority.routing.key",
message,
msg -> {
msg.getMessageProperties().setPriority(priority);
return msg;
}
);
}
@RabbitListener(queues = "priority.queue")
public void handlePriorityMessage(String message,
@Header("priority") Integer priority) {
System.out.println("Received priority " + priority + " message: " + message);
}
}
7. 性能优化和最佳实践
连接池管理
连接池配置：
@Configuration
public class RabbitConnectionConfig {
@Bean
public CachingConnectionFactory connectionFactory() {
CachingConnectionFactory factory = new CachingConnectionFactory("localhost");
factory.setUsername("admin");
factory.setPassword("admin123");
// 连接池配置
factory.setChannelCacheSize(50); // 缓存的Channel数量
factory.setConnectionCacheSize(10); // 缓存的Connection数量
factory.setChannelCheckoutTimeout(30000); // Channel获取超时时间
// 连接超时配置
factory.setConnectionTimeout(30000);
factory.setRequestedHeartBeat(60);
// 发布确认
factory.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
factory.setPublisherReturns(true);
return factory;
}
}
批量处理消息
批量发送消息：
@Service
public class BatchMessageService {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendBatchMessages(List<String> messages) {
// 使用事务批量发送
rabbitTemplate.execute(channel -> {
channel.txSelect(); // 开启事务
try {
for (String message : messages) {
channel.basicPublish(
"batch.exchange",
"batch.routing.key",
null,
message.getBytes("UTF-8")
);
}
channel.txCommit(); // 提交事务
} catch (Exception e) {
channel.txRollback(); // 回滚事务
throw new RuntimeException("Batch send failed", e);
}
return null;
});
}
// 使用发布确认批量发送
public void sendBatchWithConfirm(List<String> messages) {
rabbitTemplate.execute(channel -> {
channel.confirmSelect(); // 开启发布确认模式
for (String message : messages) {
channel.basicPublish(
"batch.exchange",
"batch.routing.key",
null,
message.getBytes("UTF-8")
);
}
// 等待所有消息确认
boolean allConfirmed = channel.waitForConfirms(5000);
if (!allConfirmed) {
throw new RuntimeException("Not all messages were confirmed");
}
return null;
});
}
}
批量消费消息：
@Component
public class BatchConsumerService {
private final List<String> messageBuffer = new ArrayList<>();
private final int BATCH_SIZE = 100;
@RabbitListener(queues = "batch.queue", ackMode = "MANUAL")
public void handleBatchMessage(
String message,
@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag,
Channel channel) throws Exception {
synchronized (messageBuffer) {
messageBuffer.add(message);
if (messageBuffer.size() >= BATCH_SIZE) {
processBatch(new ArrayList<>(messageBuffer));
messageBuffer.clear();
// 批量确认消息
channel.basicAck(deliveryTag, true);
}
}
}
private void processBatch(List<String> messages) {
System.out.println("Processing batch of " + messages.size() + " messages");
// 批量处理逻辑
for (String message : messages) {
// 处理单个消息
System.out.println("Processing: " + message);
}
}
// 定时处理剩余消息
@Scheduled(fixedDelay = 5000)
public void processRemainingMessages() {
synchronized (messageBuffer) {
if (!messageBuffer.isEmpty()) {
processBatch(new ArrayList<>(messageBuffer));
messageBuffer.clear();
}
}
}
}
消费者并发控制
并发配置：
spring:
rabbitmq:
listener:
simple:
concurrency: 5 # 初始消费者数量
max-concurrency: 20 # 最大消费者数量
prefetch: 10 # 预取消息数量
acknowledge-mode: manual
retry:
enabled: true
initial-interval: 1000
max-attempts: 3
multiplier: 2
动态并发控制：
@Component
public class DynamicConcurrencyService {
@Autowired
private SimpleRabbitListenerContainerFactory factory;
@EventListener
public void handleHighLoad(HighLoadEvent event) {
// 动态调整并发数
factory.setConcurrentConsumers(10);
factory.setMaxConcurrentConsumers(50);
}
@EventListener
public void handleLowLoad(LowLoadEvent event) {
factory.setConcurrentConsumers(2);
factory.setMaxConcurrentConsumers(10);
}
}
监控和日志记录
监控配置：
@Component
public class RabbitMQMetrics {
private final MeterRegistry meterRegistry;
private final Counter messagesSent;
private final Counter messagesReceived;
private final Timer messageProcessingTime;
public RabbitMQMetrics(MeterRegistry meterRegistry) {
this.meterRegistry = meterRegistry;
this.messagesSent = Counter.builder("rabbitmq.messages.sent")
.description("Number of messages sent")
.register(meterRegistry);
this.messagesReceived = Counter.builder("rabbitmq.messages.received")
.description("Number of messages received")
.register(meterRegistry);
this.messageProcessingTime = Timer.builder("rabbitmq.message.processing.time")
.description("Message processing time")
.register(meterRegistry);
}
public void incrementMessagesSent() {
messagesSent.increment();
}
public void incrementMessagesReceived() {
messagesReceived.increment();
}
public Timer.Sample startProcessingTimer() {
return Timer.start(meterRegistry);
}
}
日志配置：
@Component
@Slf4j
public class MessageLoggingService {
@Autowired
private RabbitMQMetrics metrics;
@RabbitListener(queues = "monitored.queue")
public void handleMonitoredMessage(String message) {
Timer.Sample sample = metrics.startProcessingTimer();
try {
log.info("Processing message: {}", message);
// 业务处理逻辑
processMessage(message);
metrics.incrementMessagesReceived();
log.info("Message processed successfully: {}", message);
} catch (Exception e) {
log.error("Error processing message: {}", message, e);
throw e;
} finally {
sample.stop(metrics.getMessageProcessingTime());
}
}
private void processMessage(String message) {
// 实际业务处理
}
}
8. 实战案例
订单处理系统
在电商系统中，订单处理涉及多个步骤：库存检查、支付处理、物流安排等。使用RabbitMQ可以实现异步处理和系统解耦。
订单消息定义：
public class OrderMessage {
private Long orderId;
private Long userId;
private List<OrderItem> items;
private BigDecimal totalAmount;
private String status;
private Date createTime;
// 构造函数、getter、setter
public OrderMessage() {}
public OrderMessage(Long orderId, Long userId, List<OrderItem> items, BigDecimal totalAmount) {
this.orderId = orderId;
this.userId = userId;
this.items = items;
this.totalAmount = totalAmount;
this.status = "CREATED";
this.createTime = new Date();
}
// getter和setter方法...
}
public class OrderItem {
private Long productId;
private String productName;
private Integer quantity;
private BigDecimal price;
// 构造函数、getter、setter...
}
订单处理配置：
@Configuration
public class OrderProcessingConfig {
public static final String ORDER_EXCHANGE = "order.exchange";
public static final String ORDER_CREATED_QUEUE = "order.created.queue";
public static final String INVENTORY_CHECK_QUEUE = "inventory.check.queue";
public static final String PAYMENT_PROCESS_QUEUE = "payment.process.queue";
public static final String SHIPPING_ARRANGE_QUEUE = "shipping.arrange.queue";
@Bean
public TopicExchange orderExchange() {
return new TopicExchange(ORDER_EXCHANGE);
}
@Bean
public Queue orderCreatedQueue() {
return QueueBuilder.durable(ORDER_CREATED_QUEUE).build();
}
@Bean
public Queue inventoryCheckQueue() {
return QueueBuilder.durable(INVENTORY_CHECK_QUEUE).build();
}
@Bean
public Queue paymentProcessQueue() {
return QueueBuilder.durable(PAYMENT_PROCESS_QUEUE).build();
}
@Bean
public Queue shippingArrangeQueue() {
return QueueBuilder.durable(SHIPPING_ARRANGE_QUEUE).build();
}
// 绑定关系
@Bean
public Binding orderCreatedBinding() {
return BindingBuilder.bind(orderCreatedQueue())
.to(orderExchange()).with("order.created");
}
@Bean
public Binding inventoryCheckBinding() {
return BindingBuilder.bind(inventoryCheckQueue())
.to(orderExchange()).with("order.inventory.check");
}
@Bean
public Binding paymentProcessBinding() {
return BindingBuilder.bind(paymentProcessQueue())
.to(orderExchange()).with("order.payment.process");
}
@Bean
public Binding shippingArrangeBinding() {
return BindingBuilder.bind(shippingArrangeQueue())
.to(orderExchange()).with("order.shipping.arrange");
}
}
订单处理服务：
@Service
@Slf4j
public class OrderProcessingService {
@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private InventoryService inventoryService;
@Autowired
private PaymentService paymentService;
@Autowired
private ShippingService shippingService;
// 创建订单
public void createOrder(OrderMessage order) {
log.info("Creating order: {}", order.getOrderId());
// 发送订单创建消息
rabbitTemplate.convertAndSend(
OrderProcessingConfig.ORDER_EXCHANGE,
"order.created",
order
);
}
// 处理订单创建
@RabbitListener(queues = OrderProcessingConfig.ORDER_CREATED_QUEUE)
public void handleOrderCreated(OrderMessage order) {
log.info("Processing created order: {}", order.getOrderId());
// 触发库存检查
rabbitTemplate.convertAndSend(
OrderProcessingConfig.ORDER_EXCHANGE,
"order.inventory.check",
order
);
}
// 处理库存检查
@RabbitListener(queues = OrderProcessingConfig.INVENTORY_CHECK_QUEUE)
public void handleInventoryCheck(OrderMessage order) {
log.info("Checking inventory for order: {}", order.getOrderId());
try {
boolean inventoryAvailable = inventoryService.checkInventory(order.getItems());
if (inventoryAvailable) {
order.setStatus("INVENTORY_CONFIRMED");
// 触发支付处理
rabbitTemplate.convertAndSend(
OrderProcessingConfig.ORDER_EXCHANGE,
"order.payment.process",
order
);
} else {
order.setStatus("INVENTORY_INSUFFICIENT");
log.warn("Insufficient inventory for order: {}", order.getOrderId());
// 发送库存不足通知
}
} catch (Exception e) {
log.error("Error checking inventory for order: {}", order.getOrderId(), e);
order.setStatus("INVENTORY_CHECK_FAILED");
}
}
// 处理支付
@RabbitListener(queues = OrderProcessingConfig.PAYMENT_PROCESS_QUEUE)
public void handlePaymentProcess(OrderMessage order) {
log.info("Processing payment for order: {}", order.getOrderId());
try {
boolean paymentSuccess = paymentService.processPayment(
order.getUserId(),
order.getTotalAmount()
);
if (paymentSuccess) {
order.setStatus("PAID");
// 触发物流安排
rabbitTemplate.convertAndSend(
OrderProcessingConfig.ORDER_EXCHANGE,
"order.shipping.arrange",
order
);
} else {
order.setStatus("PAYMENT_FAILED");
log.warn("Payment failed for order: {}", order.getOrderId());
}
} catch (Exception e) {
log.error("Error processing payment for order: {}", order.getOrderId(), e);
order.setStatus("PAYMENT_ERROR");
}
}
// 处理物流安排
@RabbitListener(queues = OrderProcessingConfig.SHIPPING_ARRANGE_QUEUE)
public void handleShippingArrange(OrderMessage order) {
log.info("Arranging shipping for order: {}", order.getOrderId());
try {
String trackingNumber = shippingService.arrangeShipping(order);
order.setStatus("SHIPPED");
log.info("Order {} shipped with tracking number: {}",
order.getOrderId(), trackingNumber);
} catch (Exception e) {
log.error("Error arranging shipping for order: {}", order.getOrderId(), e);
order.setStatus("SHIPPING_FAILED");
}
}
}
异步任务处理
对于耗时的任务，如图片处理、报表生成、邮件发送等，可以使用RabbitMQ实现异步处理。
任务处理配置：
@Configuration
public class TaskProcessingConfig {
public static final String TASK_EXCHANGE = "task.exchange";
public static final String IMAGE_PROCESS_QUEUE = "task.image.process.queue";
public static final String EMAIL_SEND_QUEUE = "task.email.send.queue";
public static final String REPORT_GENERATE_QUEUE = "task.report.generate.queue";
@Bean
public DirectExchange taskExchange() {
return new DirectExchange(TASK_EXCHANGE);
}
@Bean
public Queue imageProcessQueue() {
return QueueBuilder
.durable(IMAGE_PROCESS_QUEUE)
.withArgument("x-max-priority", 10) // 支持优先级
.build();
}
@Bean
public Queue emailSendQueue() {
return QueueBuilder.durable(EMAIL_SEND_QUEUE).build();
}
@Bean
public Queue reportGenerateQueue() {
return QueueBuilder
.durable(REPORT_GENERATE_QUEUE)
.withArgument("x-message-ttl", 300000) // 5分钟TTL
.build();
}
// 绑定关系
@Bean
public Binding imageProcessBinding() {
return BindingBuilder.bind(imageProcessQueue())
.to(taskExchange()).with("task.image.process");
}
@Bean
public Binding emailSendBinding() {
return BindingBuilder.bind(emailSendQueue())
.to(taskExchange()).with("task.email.send");
}
@Bean
public Binding reportGenerateBinding() {
return BindingBuilder.bind(reportGenerateQueue())
.to(taskExchange()).with("task.report.generate");
}
}
任务消息定义：
public class TaskMessage {
private String taskId;
private String taskType;
private Map<String, Object> parameters;
private String status;
private Date createTime;
private Date processTime;
// 构造函数、getter、setter...
}
public class ImageProcessTask extends TaskMessage {
private String imageUrl;
private String targetFormat;
private Map<String, Object> processingOptions;
// 构造函数、getter、setter...
}
public class EmailTask extends TaskMessage {
private String to;
private String subject;
private String content;
private List<String> attachments;
// 构造函数、getter、setter...
}
任务处理服务：
@Service
@Slf4j
public class TaskProcessingService {
@Autowired
private RabbitTemplate rabbitTemplate;
// 提交图片处理任务
public void submitImageProcessTask(ImageProcessTask task, int priority) {
rabbitTemplate.convertAndSend(
TaskProcessingConfig.TASK_EXCHANGE,
"task.image.process",
task,
message -> {
message.getMessageProperties().setPriority(priority);
return message;
}
);
log.info("Submitted image process task: {}", task.getTaskId());
}
// 处理图片处理任务
@RabbitListener(queues = TaskProcessingConfig.IMAGE_PROCESS_QUEUE,
concurrency = "2-10") // 动态并发
public void handleImageProcessTask(ImageProcessTask task) {
log.info("Processing image task: {}", task.getTaskId());
try {
task.setStatus("PROCESSING");
task.setProcessTime(new Date());
// 图片处理逻辑
processImage(task);
task.setStatus("COMPLETED");
log.info("Image task completed: {}", task.getTaskId());
} catch (Exception e) {
task.setStatus("FAILED");
log.error("Image task failed: {}", task.getTaskId(), e);
}
}
// 处理邮件发送任务
@RabbitListener(queues = TaskProcessingConfig.EMAIL_SEND_QUEUE)
public void handleEmailSendTask(EmailTask task) {
log.info("Sending email task: {}", task.getTaskId());
try {
sendEmail(task);
task.setStatus("SENT");
log.info("Email sent successfully: {}", task.getTaskId());
} catch (Exception e) {
task.setStatus("FAILED");
log.error("Email send failed: {}", task.getTaskId(), e);
// 重试逻辑或进入死信队列
throw new AmqpRejectAndDontRequeueException("Email send failed", e);
}
}
private void processImage(ImageProcessTask task) {
// 实现图片处理逻辑
log.info("Processing image: {} -> {}", task.getImageUrl(), task.getTargetFormat());
// 模拟处理时间
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
private void sendEmail(EmailTask task) {
// 实现邮件发送逻辑
log.info("Sending email to: {}, subject: {}", task.getTo(), task.getSubject());
// 模拟发送时间
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
微服务间通信
在微服务架构中，RabbitMQ可以作为服务间异步通信的消息总线。
事件驱动架构配置：
@Configuration
public class EventDrivenConfig {
public static final String EVENT_EXCHANGE = "event.exchange";
public static final String USER_SERVICE_QUEUE = "user.service.queue";
public static final String ORDER_SERVICE_QUEUE = "order.service.queue";
public static final String NOTIFICATION_SERVICE_QUEUE = "notification.service.queue";
@Bean
public TopicExchange eventExchange() {
return new TopicExchange(EVENT_EXCHANGE);
}
@Bean
public Queue userServiceQueue() {
return QueueBuilder.durable(USER_SERVICE_QUEUE).build();
}
@Bean
public Queue orderServiceQueue() {
return QueueBuilder.durable(ORDER_SERVICE_QUEUE).build();
}
@Bean
public Queue notificationServiceQueue() {
return QueueBuilder.durable(NOTIFICATION_SERVICE_QUEUE).build();
}
// 用户服务监听用户相关事件
@Bean
public Binding userServiceBinding() {
return BindingBuilder.bind(userServiceQueue())
.to(eventExchange()).with("user.*");
}
// 订单服务监听订单和用户事件
@Bean
public Binding orderServiceBinding1() {
return BindingBuilder.bind(orderServiceQueue())
.to(eventExchange()).with("order.*");
}
@Bean
public Binding orderServiceBinding2() {
return BindingBuilder.bind(orderServiceQueue())
.to(eventExchange()).with("user.registered");
}
// 通知服务监听所有事件
@Bean
public Binding notificationServiceBinding() {
return BindingBuilder.bind(notificationServiceQueue())
.to(eventExchange()).with("*.*");
}
}
事件处理服务：
@Service
@Slf4j
public class EventService {
@Autowired
private RabbitTemplate rabbitTemplate;
// 发布用户注册事件
public void publishUserRegisteredEvent(UserRegisteredEvent event) {
rabbitTemplate.convertAndSend(
EventDrivenConfig.EVENT_EXCHANGE,
"user.registered",
event
);
log.info("Published user registered event: {}", event.getUserId());
}
// 发布订单创建事件
public void publishOrderCreatedEvent(OrderCreatedEvent event) {
rabbitTemplate.convertAndSend(
EventDrivenConfig.EVENT_EXCHANGE,
"order.created",
event
);
log.info("Published order created event: {}", event.getOrderId());
}
// 用户服务处理用户事件
@RabbitListener(queues = EventDrivenConfig.USER_SERVICE_QUEUE)
public void handleUserEvents(
@Payload Object event,
@Header("amqp_receivedRoutingKey") String routingKey) {
log.info("User service received event: {} with routing key: {}",
event.getClass().getSimpleName(), routingKey);
switch (routingKey) {
case "user.registered":
handleUserRegistered((UserRegisteredEvent) event);
break;
case "user.updated":
handleUserUpdated((UserUpdatedEvent) event);
break;
default:
log.warn("Unknown user event: {}", routingKey);
}
}
// 订单服务处理订单和用户事件
@RabbitListener(queues = EventDrivenConfig.ORDER_SERVICE_QUEUE)
public void handleOrderEvents(
@Payload Object event,
@Header("amqp_receivedRoutingKey") String routingKey) {
log.info("Order service received event: {} with routing key: {}",
event.getClass().getSimpleName(), routingKey);
switch (routingKey) {
case "user.registered":
// 为新用户创建优惠券
createWelcomeCoupon((UserRegisteredEvent) event);
break;
case "order.created":
handleOrderCreated((OrderCreatedEvent) event);
break;
case "order.paid":
handleOrderPaid((OrderPaidEvent) event);
break;
default:
log.warn("Unknown order event: {}", routingKey);
}
}
// 通知服务处理所有事件
@RabbitListener(queues = EventDrivenConfig.NOTIFICATION_SERVICE_QUEUE)
public void handleNotificationEvents(
@Payload Object event,
@Header("amqp_receivedRoutingKey") String routingKey) {
log.info("Notification service received event: {} with routing key: {}",
event.getClass().getSimpleName(), routingKey);
// 根据事件类型发送不同的通知
switch (routingKey) {
case "user.registered":
sendWelcomeNotification((UserRegisteredEvent) event);
break;
case "order.created":
sendOrderConfirmationNotification((OrderCreatedEvent) event);
break;
case "order.shipped":
sendShippingNotification((OrderShippedEvent) event);
break;
default:
log.debug("No notification needed for event: {}", routingKey);
}
}
private void handleUserRegistered(UserRegisteredEvent event) {
log.info("Handling user registration: {}", event.getUserId());
// 用户服务内部处理逻辑
}
private void handleUserUpdated(UserUpdatedEvent event) {
log.info("Handling user update: {}", event.getUserId());
// 用户更新处理逻辑
}
private void createWelcomeCoupon(UserRegisteredEvent event) {
log.info("Creating welcome coupon for user: {}", event.getUserId());
// 创建新用户优惠券逻辑
}
private void handleOrderCreated(OrderCreatedEvent event) {
log.info("Handling order creation: {}", event.getOrderId());
// 订单创建处理逻辑
}
private void handleOrderPaid(OrderPaidEvent event) {
log.info("Handling order payment: {}", event.getOrderId());
// 订单支付处理逻辑
}
private void sendWelcomeNotification(UserRegisteredEvent event) {
log.info("Sending welcome notification to user: {}", event.getUserId());
// 发送欢迎通知逻辑
}
private void sendOrderConfirmationNotification(OrderCreatedEvent event) {
log.info("Sending order confirmation notification for order: {}", event.getOrderId());
// 发送订单确认通知逻辑
}
private void sendShippingNotification(OrderShippedEvent event) {
log.info("Sending shipping notification for order: {}", event.getOrderId());
// 发送发货通知逻辑
}
}
// 事件类定义
public class UserRegisteredEvent {
private Long userId;
private String email;
private String username;
private Date registrationTime;
// 构造函数、getter、setter...
}
public class UserUpdatedEvent {
private Long userId;
private Map<String, Object> updatedFields;
private Date updateTime;
// 构造函数、getter、setter...
}
public class OrderCreatedEvent {
private Long orderId;
private Long userId;
private BigDecimal totalAmount;
private Date createTime;
// 构造函数、getter、setter...
}
public class OrderPaidEvent {
private Long orderId;
private Long userId;
private BigDecimal paidAmount;
private String paymentMethod;
private Date paidTime;
// 构造函数、getter、setter...
}
public class OrderShippedEvent {
private Long orderId;
private String trackingNumber;
private String shippingCompany;
private Date shippedTime;
// 构造函数、getter、setter...
}
总结
通过本文的详细介绍，我们深入了解了RabbitMQ在Java应用中的使用。从基础概念到高级特性，从简单的点对点通信到复杂的事件驱动架构，RabbitMQ都能提供强大的支持。
关键要点回顾
基础概念掌握：理解Exchange、Queue、Binding等核心概念是使用RabbitMQ的基础。
消息模式选择：根据业务场景选择合适的消息模式，如简单队列用于任务分发，发布订阅用于广播通知。
Spring Boot集成：使用Spring AMQP可以大大简化RabbitMQ的使用，提供了丰富的注解和配置选项。
可靠性保证：通过消息确认、死信队列、消息持久化等机制确保消息的可靠传递。
性能优化：合理配置连接池、批量处理、并发控制等参数来优化系统性能。
监控运维：建立完善的监控和日志体系，及时发现和解决问题。
最佳实践建议
- 设计原则：遵循单一职责原则，每个队列处理特定类型的消息
- 错误处理：建立完善的错误处理和重试机制
- 资源管理：合理管理连接和通道资源，避免资源泄露
- 安全考虑：使用适当的认证和授权机制保护消息安全
- 测试策略：编写完善的单元测试和集成测试
RabbitMQ作为一个成熟的消息中间件，在微服务架构、异步处理、系统解耦等场景中发挥着重要作用。掌握其使用方法和最佳实践，将有助于构建更加可靠、可扩展的分布式系统。
本文涵盖了RabbitMQ在Java中的主要使用场景和实践方法。在实际应用中，还需要根据具体的业务需求和系统架构进行适当的调整和优化。