当前位置：首页 > news >正文

RabbitMQ全面详解：从核心概念到企业级应用

news 2025/10/15 6:44:50

1 RabbitMQ简介与核心架构

1.1 什么是RabbitMQ？

RabbitMQ是一个开源的消息代理中间件，实现了高级消息队列协议（AMQP）。它使用Erlang语言开发，以其高可靠性、灵活的路由机制和易用性而闻名。RabbitMQ在分布式系统中扮演着消息中转站的角色，用于实现系统间的异步通信、服务解耦和流量削峰。

1.2 核心组件与架构

理解RabbitMQ的架构是正确使用它的基础，其核心包含以下几个组件：

Producer（生产者）：发送消息的应用程序，将消息发送到Exchange。
Exchange（交换机）：接收生产者发送的消息，并根据路由规则将消息分发到相应的队列。RabbitMQ支持多种交换机类型，每种类型对应不同的消息路由机制。
Queue（队列）：存储消息的缓冲区，是消息的最终目的地。消费者从队列中获取消息进行处理。
Consumer（消费者）：从队列接收并处理消息的应用程序。
Binding（绑定）：连接Exchange和Queue的规则，定义了消息应该如何从Exchange路由到Queue。
Channel（信道）：在TCP连接内创建的逻辑连接，避免频繁创建和销毁TCP连接的开销。
Virtual Host（虚拟主机）：提供命名空间和环境，将Exchange、Queue等资源进行逻辑分组。

1.3 RabbitMQ的特点与优势

高可靠性：支持消息持久化、传输确认、镜像队列等高可用特性。
灵活的路由：通过多种Exchange类型和Binding规则，实现复杂的消息路由逻辑。
集群支持：支持集群部署，实现水平扩展和高可用性。
多协议支持：除了AMQP，还支持STOMP、MQTT等多种消息协议。
丰富的客户端支持：提供Java、Python、.NET等多种语言的客户端库。

表：RabbitMQ核心组件功能总结

组件	角色定位	关键特性
Producer	消息生产者	创建并发送消息到Exchange
Exchange	消息路由中心	根据类型和规则将消息分发到相应队列
Queue	消息存储区	缓冲存储消息，供消费者消费
Consumer	消息消费者	从队列获取并处理消息
Binding	路由规则定义	建立Exchange与Queue的关联关系

2 环境搭建与配置

2.1 安装RabbitMQ

以下是基于不同环境的RabbitMQ安装方法：

使用Docker安装（推荐）：

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 \-e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=secret \rabbitmq:3-management

这种方法简单快捷，适合开发和测试环境。

在Ubuntu上安装：

sudo apt-get update
sudo apt-get install -y erlang rabbitmq-server
sudo systemctl start rabbitmq-server
sudo systemctl enable rabbitmq-server

安装前需确保已安装Erlang，因为RabbitMQ是用Erlang编写的。

在CentOS/RHEL上安装：

# 添加Erlang Solutions仓库
curl -s https://packagecloud.io/install/repositories/rabbitmq/erlang/script.rpm.sh | sudo bash
# 安装Erlang
sudo dnf install -y erlang
# 添加RabbitMQ仓库并安装
sudo rpm --import https://github.com/rabbitmq/signing-keys/releases/download/2.0/rabbitmq-release-signing-key.asc
sudo dnf install -y rabbitmq-server
# 启动服务
sudo systemctl enable rabbitmq-server
sudo systemctl start rabbitmq-server

2.2 基本配置与管理

启用管理插件：
RabbitMQ提供了一个Web管理界面，方便监控和管理：

sudo rabbitmq-plugins enable rabbitmq_management

启用后可通过http://服务器IP:15672 访问，默认用户名和密码为guest/guest（默认仅限本地访问）。

创建用户和虚拟主机：
为安全考虑，建议创建专用用户和虚拟主机：

# 添加用户
sudo rabbitmqctl add_user myuser mypassword
# 设置用户角色
sudo rabbitmqctl set_user_tags myuser administrator
# 创建虚拟主机
sudo rabbitmqctl add_vhost myvhost
# 设置权限
sudo rabbitmqctl set_permissions -p myvhost myuser ".*" ".*" ".*"

2.3 关键配置详解

RabbitMQ的主要配置文件通常位于/etc/rabbitmq/rabbitmq.conf，常见配置项包括：

# 监听端口配置
listeners.tcp.default = 5672
# 管理界面端口
management.tcp.port = 15672
# 默认用户权限（允许远程访问）
loopback_users = none
# 内存阈值
vm_memory_high_watermark.absolute = 2GB

3 RabbitMQ工作模式详解

RabbitMQ支持多种工作模式，每种模式适用于不同的业务场景。下面将详细介绍这些模式及其Java实现。

3.1 简单模式（Simple Queue）

最简单的消息队列模式，一个生产者直接向队列发送消息，一个消费者从队列接收消息。

Java生产者示例：

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;public class SimpleProducer {private final static String QUEUE_NAME = "hello";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 声明队列，持久化设置为truechannel.queueDeclare(QUEUE_NAME, true, false, false, null);String message = "Hello RabbitMQ!";// 发送消息，设置消息持久化channel.basicPublish("", QUEUE_NAME, new AMQP.BasicProperties.Builder().deliveryMode(2) // 持久化消息.build(),message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + message + "'");}}
}

Java消费者示例：

import com.rabbitmq.client.*;public class SimpleConsumer {private final static String QUEUE_NAME = "hello";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 声明队列，确保生产者先创建队列channel.queueDeclare(QUEUE_NAME, true, false, false, null);System.out.println(" [*] Waiting for messages. To exit press CTRL+C");// 创建消费者DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");// 手动确认消息channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);};// 消费消息，关闭自动确认channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> { });}
}

3.2 工作队列模式（Work Queue）

一个生产者，一个队列，多个消费者竞争消费消息，适用于任务分发和负载均衡。

Java生产者示例（发送耗时任务）：

public class WorkQueueProducer {private final static String TASK_QUEUE_NAME = "task_queue";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 声明持久化队列channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);// 发送10个任务for (int i = 1; i <= 10; i++) {String message = "Task " + i + " [" + ".".repeat(i) + "]";channel.basicPublish("", TASK_QUEUE_NAME, new AMQP.BasicProperties.Builder().deliveryMode(2).build(),message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + message + "'");}}}
}

Java消费者示例（公平分发）：

public class WorkQueueConsumer {private final static String TASK_QUEUE_NAME = "task_queue";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);System.out.println(" [*] Waiting for messages. To exit press CTRL+C");// 每次只预取一条消息，确保公平分发channel.basicQos(1);DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");try {// 模拟耗时任务doWork(message);} finally {// 任务完成后手动确认channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);System.out.println(" [x] Done");}};// 关闭自动确认，手动控制消息确认channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, consumerTag -> { });}private static void doWork(String task) {for (char ch : task.toCharArray()) {if (ch == '.') {try {Thread.sleep(1000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}}}}
}

3.3 发布/订阅模式（Publish/Subscribe）

使用Fanout交换机将消息广播到所有绑定的队列，实现一对多通信。

Java生产者示例：

public class PublishSubscribeProducer {private static final String EXCHANGE_NAME = "logs";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 声明fanout交换机channel.exchangeDeclare(EXCHANGE_NAME, "fanout");// 发送广播消息for (int i = 1; i <= 5; i++) {String message = "Log message " + i;channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());System.out.println(" [x] Sent '" + message + "'");}}}
}

Java消费者示例：

public class PublishSubscribeConsumer {private static final String EXCHANGE_NAME = "logs";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 声明交换机channel.exchangeDeclare(EXCHANGE_NAME, "fanout");// 创建临时队列（断开连接后自动删除）String queueName = channel.queueDeclare().getQueue();// 将队列绑定到交换机channel.queueBind(queueName, EXCHANGE_NAME, "");System.out.println(" [*] Waiting for messages. To exit press CTRL+C");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received '" + message + "'");};channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });}
}

3.4 路由模式（Routing）

使用Direct交换机根据路由键精确匹配，将消息路由到特定的队列。

Java生产者示例：

public class RoutingProducer {private static final String EXCHANGE_NAME = "direct_logs";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 声明direct交换机channel.exchangeDeclare(EXCHANGE_NAME, "direct");// 发送不同级别的日志消息String[] severities = {"info", "warning", "error"};for (String severity : severities) {String message = severity.toUpperCase() + " log message";channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());System.out.println(" [x] Sent '" + severity + "':'" + message + "'");}}}
}

Java消费者示例（只接收error级别日志）：

public class RoutingConsumer {private static final String EXCHANGE_NAME = "direct_logs";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 声明交换机channel.exchangeDeclare(EXCHANGE_NAME, "direct");// 创建临时队列String queueName = channel.queueDeclare().getQueue();// 只绑定error级别的日志channel.queueBind(queueName, EXCHANGE_NAME, "error");System.out.println(" [*] Waiting for error logs. To exit press CTRL+C");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");String routingKey = delivery.getEnvelope().getRoutingKey();System.out.println(" [x] Received '" + routingKey + "':'" + message + "'");};channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });}
}

3.5 主题模式（Topics）

使用Topic交换机支持通配符匹配，实现更灵活的消息路由。

Java生产者示例：

public class TopicProducer {private static final String EXCHANGE_NAME = "topic_logs";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 声明topic交换机channel.exchangeDeclare(EXCHANGE_NAME, "topic");// 发送不同主题的消息String[] routingKeys = {"order.created", "order.paid", "user.created", "user.deleted"};for (String routingKey : routingKeys) {String message = "Message for " + routingKey;channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");}}}
}

Java消费者示例（使用通配符）：

public class TopicConsumer {private static final String EXCHANGE_NAME = "topic_logs";public static void main(String[] argv) throws Exception {if (argv.length < 1) {System.err.println("Usage: TopicConsumer [binding_key]...");System.exit(1);}ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 声明交换机channel.exchangeDeclare(EXCHANGE_NAME, "topic");String queueName = channel.queueDeclare().getQueue();// 绑定所有传入的路由键模式for (String bindingKey : argv) {channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);System.out.println(" [*] Binding key: " + bindingKey);}System.out.println(" [*] Waiting for messages. To exit press CTRL+C");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");String routingKey = delivery.getEnvelope().getRoutingKey();System.out.println(" [x] Received '" + routingKey + "':'" + message + "'");};channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });}
}

使用示例：java TopicConsumer "order.*" "user.*"

表：RabbitMQ工作模式对比

工作模式	交换机类型	路由方式	典型应用场景
简单模式	默认交换机	直接指定队列名	简单的点对点通信
工作队列模式	默认交换机	直接指定队列名	任务分发、负载均衡
发布/订阅模式	Fanout	广播到所有绑定队列	系统通知、事件广播
路由模式	Direct	精确匹配路由键	日志分级处理、条件路由
主题模式	Topic	通配符匹配路由键	复杂的消息过滤、多条件路由

4 RabbitMQ高级特性

4.1 消息确认机制（Acknowledgement）

RabbitMQ提供两种消息确认方式，确保消息可靠消费。

自动确认：

// 消息发送后立即确认，有丢失风险
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });

手动确认（推荐）：

// 关闭自动确认
channel.basicConsume(queueName, false, deliverCallback, consumerTag -> { });// 在处理完消息后手动确认
DeliverCallback deliverCallback = (consumerTag, delivery) -> {try {String message = new String(delivery.getBody(), "UTF-8");// 处理业务逻辑processMessage(message);// 处理成功，确认消息channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);} catch (Exception e) {// 处理失败，拒绝消息（可设置重新入队）channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);}
};

4.2 消息持久化

通过队列持久化、消息持久化和交换机持久化，确保RabbitMQ重启后消息不丢失。

// 1. 队列持久化
boolean durable = true;
channel.queueDeclare("task_queue", durable, false, false, null);// 2. 消息持久化
channel.basicPublish("", "task_queue",new AMQP.BasicProperties.Builder().deliveryMode(2) // 持久化消息.build(),message.getBytes());// 3. 交换机持久化（如果需要）
channel.exchangeDeclare("logs", "fanout", true);

4.3 预取计数（Prefetch Count）

通过设置预取数量，实现公平分发，避免某个消费者过载。

// 每次只预取一条消息，确保公平分发
int prefetchCount = 1;
channel.basicQos(prefetchCount);

4.4 消息TTL（Time-To-Live）

设置消息或队列的过期时间，自动清理过期消息。

队列级别TTL：

Map<String, Object> args = new HashMap<>();
args.put("x-message-ttl", 60000); // 60秒
channel.queueDeclare("my_queue", false, false, false, args);

消息级别TTL：

channel.basicPublish("", "queue_name",new AMQP.BasicProperties.Builder().expiration("60000") // 60秒.build(),message.getBytes());

4.5 死信队列（Dead Letter Exchange）

处理无法被正常消费的消息，实现异常消息处理机制。

// 创建死信交换机
channel.exchangeDeclare("dlx", "direct");// 创建死信队列
channel.queueDeclare("dl_queue", true, false, false, null);
channel.queueBind("dl_queue", "dlx", "dl_routing_key");// 创建正常队列，设置死信参数
Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange", "dlx");
args.put("x-dead-letter-routing-key", "dl_routing_key");
channel.queueDeclare("normal_queue", true, false, false, args);

5 Spring Boot集成RabbitMQ

5.1 添加依赖和配置

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

application.yml配置：

spring:rabbitmq:host: localhostport: 5672username: guestpassword: guest# 消息确认配置publisher-confirm-type: correlatedpublisher-returns: truetemplate:mandatory: truelistener:simple:acknowledge-mode: manual  # 手动确认prefetch: 1  # 每次预取1条消息

5.2 配置类

@Configuration
public class RabbitMQConfig {// 定义队列@Beanpublic Queue orderQueue() {return new Queue("order.queue", true); // 持久化队列}// 定义交换机@Beanpublic DirectExchange orderExchange() {return new DirectExchange("order.exchange");}// 绑定队列和交换机@Beanpublic Binding binding(Queue orderQueue, DirectExchange orderExchange) {return BindingBuilder.bind(orderQueue).to(orderExchange).with("order.routingkey");}// 配置JSON消息转换器@Beanpublic MessageConverter jsonMessageConverter() {return new Jackson2JsonMessageConverter();}
}

5.3 生产者示例

@Component
public class OrderMessageProducer {@Autowiredprivate AmqpTemplate rabbitTemplate;public void sendOrderCreated(Order order) {// 发送普通消息rabbitTemplate.convertAndSend("order.exchange", "order.routingkey", order);// 发送带TTL的延迟消息Message message = MessageBuilder.withBody(order.toString().getBytes()).setExpiration("60000") // 60秒后过期.build();rabbitTemplate.send("order.exchange", "order.routingkey", message);}
}

5.4 消费者示例

@Component
public class OrderMessageConsumer {@RabbitListener(queues = "order.queue")public void handleOrderMessage(Order order, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {try {// 处理订单业务逻辑processOrder(order);// 手动确认消息channel.basicAck(tag, false);} catch (Exception e) {// 处理失败，拒绝消息（不重新入队）try {channel.basicReject(tag, false);} catch (IOException ex) {ex.printStackTrace();}}}private void processOrder(Order order) {// 订单处理逻辑System.out.println("Processing order: " + order.getId());}
}

6 实战案例：电商系统应用

下面通过一个完整的电商订单案例展示RabbitMQ在实际项目中的应用。

6.1 系统架构设计

用户下单 → 订单服务 → 订单相关Exchange → 库存服务、物流服务、通知服务

6.2 订单消息生产者

@Service
public class OrderService {@Autowiredprivate AmqpTemplate rabbitTemplate;public void createOrder(Order order) {// 1. 保存订单到数据库boolean saveSuccess = orderRepository.save(order);if (saveSuccess) {// 2. 发送订单创建消息到不同服务sendOrderCreatedEvent(order);// 3. 发送延迟消息检查支付超时sendOrderTimeoutCheck(order);}}private void sendOrderCreatedEvent(Order order) {// 发送到库存服务（路由模式）rabbitTemplate.convertAndSend("order.exchange", "order.created.inventory", order);// 发送到物流服务rabbitTemplate.convertAndSend("order.exchange", "order.created.logistics", order);// 发送到通知服务（广播模式）rabbitTemplate.convertAndSend("notification.exchange", "", order);}private void sendOrderTimeoutCheck(Order order) {// 创建延迟消息，30分钟未支付自动取消Message message = MessageBuilder.withBody(order.getId().getBytes()).setExpiration("1800000") // 30分钟.build();rabbitTemplate.send("order.delay.exchange", "order.timeout", message);}
}

6.3 库存服务消费者

@Service
public class InventoryService {@RabbitListener(bindings = @QueueBinding(value = @Queue(value = "inventory.queue", durable = "true"),exchange = @Exchange(value = "order.exchange", type = "topic"),key = "order.created.inventory"))public void handleOrderCreated(Order order, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {try {// 扣减库存boolean deductSuccess = inventoryRepository.deductStock(order);if (deductSuccess) {System.out.println("库存扣减成功，订单号：" + order.getId());channel.basicAck(tag, false);} else {// 库存不足，拒绝消息并不重新入队channel.basicReject(tag, false);// 发送库存不足通知sendInventoryShortageNotification(order);}} catch (Exception e) {// 记录日志并重新入队try {channel.basicNack(tag, false, true);} catch (IOException ex) {ex.printStackTrace();}}}
}

6.4 订单超时检查消费者

@Service
public class OrderTimeoutService {@RabbitListener(queues = "order.timeout.queue")public void checkOrderTimeout(String orderId, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {try {Order order = orderRepository.findById(orderId);if (order != null && order.getStatus() == OrderStatus.UNPAID) {// 超时未支付，自动取消订单order.setStatus(OrderStatus.CANCELLED);orderRepository.update(order);// 恢复库存restoreInventory(order);System.out.println("订单超时取消：" + orderId);}channel.basicAck(tag, false);} catch (Exception e) {try {channel.basicNack(tag, false, true);} catch (IOException ex) {ex.printStackTrace();}}}
}

7 集群与高可用配置

7.1 RabbitMQ集群搭建

# 在节点2上执行，加入集群
rabbitmqctl stop_app
rabbitmqctl join_cluster rabbit@node1
rabbitmqctl start_app# 查看集群状态
rabbitmqctl cluster_status

7.2 镜像队列配置

通过策略实现队列的跨节点复制，确保高可用性。

# 设置镜像队列策略
rabbitmqctl set_policy ha-all "^ha\." '{"ha-mode":"all"}'

Java配置方式：

@Bean
public Queue mirroredQueue() {Map<String, Object> args = new HashMap<>();args.put("x-ha-policy", "all"); // 镜像到所有节点return new Queue("ha.queue", true, false, false, args);
}

7.3 负载均衡配置

结合HAProxy或Nginx实现客户端负载均衡：

# haproxy.cfg 配置示例
listen rabbitmq_clusterbind *:5670mode tcpbalance roundrobinserver node1 192.168.1.101:5672 checkserver node2 192.168.1.102:5672 checkserver node3 192.168.1.103:5672 check

8 监控与故障排查

8.1 关键监控指标

队列深度：监控消息堆积情况
消息速率：生产/消费速率是否平衡
连接数：客户端连接数量
内存和磁盘使用：避免资源耗尽

8.2 常见问题与解决方案

1. 消息堆积

// 解决方案：增加消费者数量或提高处理能力
@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory() {SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();factory.setConcurrentConsumers(10); // 增加并发消费者factory.setMaxConcurrentConsumers(20);return factory;
}

2. 消息重复消费
实现消费幂等性：

@Service
public class IdempotentConsumer {@Autowiredprivate MessageLogRepository messageLogRepository;@RabbitListener(queues = "business.queue")public void processMessage(Order order, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag,@Header(AmqpHeaders.MESSAGE_ID) String messageId) {// 检查消息是否已处理if (messageLogRepository.existsByMessageId(messageId)) {channel.basicAck(tag, false); // 已处理，直接确认return;}// 处理业务逻辑processBusiness(order);// 记录已处理消息messageLogRepository.save(new MessageLog(messageId, System.currentTimeMillis()));channel.basicAck(tag, false);}
}

3. 连接故障恢复

@Bean
public ConnectionFactory connectionFactory() {CachingConnectionFactory connectionFactory = new CachingConnectionFactory();connectionFactory.setHost("localhost");connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");// 开启自动重连connectionFactory.setRequestedHeartBeat(30);connectionFactory.setConnectionTimeout(30000);return connectionFactory;
}