国内设计师个人网站,研发流程,响应式网站内容布局,asp 网站管理工具1. RabbitMQ 基础概念
1.1 消息处理流程与组件配合
Producer#xff08;生产者#xff09; 发送消息。消息先发送到 Exchange#xff08;交换机#xff09;#xff0c;而不是直接到队列。Exchange#xff08;交换机#xff09; 接收到消息后#xff0c;根据 Routing …1. RabbitMQ 基础概念
1.1 消息处理流程与组件配合
Producer生产者 发送消息。消息先发送到 Exchange交换机而不是直接到队列。Exchange交换机 接收到消息后根据 Routing Key路由键 和 Binding绑定规则决定将消息发送到哪些 Queue队列。Queue队列 存储消息等待 Consumer消费者 消费。Consumer消费者 从队列中接收并处理消息。
Producer生产者
作用负责发送消息到 RabbitMQ 的入口指定消息的 Exchange 和 Routing Key。
关键点
Producer 只需要知道 Exchange 和 Routing Key不关心队列。Producer 不直接与队列交互消息的路由和存储由 Exchange 和 Binding 决定。
代码示例
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;Service
public class MessageProducer {Autowiredprivate RabbitTemplate rabbitTemplate;public void sendMessage(String exchange, String routingKey, String message) {rabbitTemplate.convertAndSend(exchange, routingKey, message);System.out.println(Sent message: message);}
}
调用示例
producer.sendMessage(direct-exchange, key1, Hello RabbitMQ);direct-exchange目标交换机。key1消息的路由键。 Exchange交换机
作用接收来自 Producer 的消息并根据 Routing Key 和 Binding 的配置决定将消息发送到哪些队列。 Exchange 通常需要手动注册为 Bean。 RabbitMQ 的 Exchange 是通过名称来标识的。 在 Spring Boot 中您通过 Bean 方法注册 Exchange 时实际上是将 Exchange 的名称和类型绑定到 RabbitMQ 服务器。 发送消息时RabbitMQ 客户端会根据 Exchange 的名称找到对应的 Exchange并根据 Routing Key 将消息路由到队列。 类型 Direct Exchange精确匹配 Routing Key。消息的 Routing Key 必须与 Binding 的 Routing Key 完全一致。 Topic Exchange支持通配符匹配。例如with(key.*) 可以匹配 key.1、key.2 等。 Fanout Exchange忽略 Routing Key消息会被广播到所有绑定的队列。 Headers Exchange忽略 Routing Key根据消息头属性匹配。
代码示例定义交换机
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.core.HeadersExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;Configuration
public class ExchangeConfig {Beanpublic DirectExchange directExchange() {return new DirectExchange(direct-exchange);}Beanpublic FanoutExchange fanoutExchange() {return new FanoutExchange(fanout-exchange);}Beanpublic TopicExchange topicExchange() {return new TopicExchange(topic-exchange);}Beanpublic HeadersExchange headersExchange() {return new HeadersExchange(headers-exchange);}
}Queue队列
作用消息的存储容器等待消费者从中取出消息进行处理。 Queue 也需要手动注册为 Bean。Spring Boot 不会自动注册队列因为队列的名称和属性如是否持久化、是否排他等需要根据业务需求进行配置。 关键点
消息会保存在队列中直到被消费。队列可以是持久化的重启 RabbitMQ 后消息仍然存在或非持久化的。
代码示例定义队列
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;Configuration
public class QueueConfig {Beanpublic Queue demoQueue() {return new Queue(demo-queue, true); // 持久化队列}
}Routing Key路由键
作用决定消息如何从交换机路由到队列。
关键点
Routing Key 由 Producer 指定。在 Direct 和 Topic 类型的 Exchange 中Routing Key 决定队列是否接收消息。 Binding绑定
作用将队列与交换机连接并定义路由规则。关键点 Binding 定义了队列接受消息的条件。结合 Routing Key 和交换机类型共同决定消息的路由方式。
代码示例定义绑定
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;Configuration
public class BindingConfig {Beanpublic Binding binding(Queue demoQueue, DirectExchange directExchange) {return BindingBuilder.bind(demoQueue).to(directExchange).with(key1);}
}with(key1) 的作用是 指定 Binding 的 Routing Key。它的含义是 当消息发送到 Exchange 时Exchange 会根据消息的 Routing Key 和 Binding 的 Routing Key 进行匹配。 如果匹配成功消息会被路由到对应的队列如果匹配失败消息会被丢弃或进入死信队列如果有配置。 Consumer消费者
作用从队列中接收并处理消息。
关键点
消费者与队列直接关联。多个消费者可以监听同一队列实现负载均衡。
代码示例
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;Service
public class Consumer {RabbitListener(queues demo-queue)public void receiveMessage(String message) {System.out.println(Received message: message);}
}1.2 RabbitMQ 消息传输模型
点对点模型
定义消息从生产者发送到队列由消费者从队列中接收消息只能被一个消费者消费。
实现
使用默认交换机空字符串 。直接将消息发送到队列。
代码示例
rabbitTemplate.convertAndSend(, demo-queue, Point-to-Point Message);发布订阅模型
定义生产者将消息发送到 Fanout 类型的交换机消息会广播到所有绑定的队列。
实现
不需要 Routing Key。所有绑定到 Fanout 交换机的队列都会接收消息。
代码示例
rabbitTemplate.convertAndSend(fanout-exchange, , Fanout Message);路由模型
定义生产者将消息发送到 Direct 类型的交换机根据 Routing Key 精确匹配队列。
实现
队列通过 Binding 绑定到交换机时指定 Routing Key。消息的 Routing Key 必须与 Binding 的 Routing Key 一致。
代码示例
rabbitTemplate.convertAndSend(direct-exchange, key1, Routing Message);2. 环境准备
2.1 安装与配置 RabbitMQ
下载 Docker 访问 Docker 官方网站Docker: Accelerated Container Application Development。 根据您的操作系统Windows、macOS 或 Linux下载并安装 Docker Desktop。
启动 Docker 安装完成后启动 Docker Desktop。 确保 Docker 正在运行任务栏或菜单栏中可以看到 Docker 图标。
使用 Docker 快速部署 RabbitMQ
Docker 是部署 RabbitMQ 的最简单方式。通过以下命令您可以快速启动一个 RabbitMQ 容器
docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:management
参数说明 -d以后台模式运行容器。 --name rabbitmq为容器指定名称rabbitmq。 -p 5672:5672将容器的 5672 端口映射到主机的 5672 端口RabbitMQ 的消息通信端口。 -p 15672:15672将容器的 15672 端口映射到主机的 15672 端口RabbitMQ 管理插件的 Web 界面端口。 rabbitmq:management使用带有管理插件的 RabbitMQ 镜像。
验证 RabbitMQ 是否运行
运行以下命令查看容器是否正常运行
docker ps
如果看到 rabbitmq 容器正在运行说明 RabbitMQ 已成功启动。 2.2 使用 RabbitMQ 管理插件
RabbitMQ 提供了一个 Web 管理界面方便您监控和管理 RabbitMQ。
访问管理界面 打开浏览器访问 http://localhost:15672。 使用默认用户名和密码登录 用户名guest 密码guest
管理界面功能 Overview查看 RabbitMQ 的整体状态如连接数、队列数、消息速率等。 Connections查看当前连接到 RabbitMQ 的客户端。 Channels查看当前打开的通道。 Exchanges查看和管理 Exchange。 Queues查看和管理 Queue。 Admin管理用户和权限。
2.3 用户与权限配置
默认情况下RabbitMQ 只有一个用户 guest密码也是 guest。为了安全性和权限管理建议创建新用户并分配权限。
1. 创建新用户
在 RabbitMQ 管理界面中 点击顶部导航栏的 Admin。 在用户列表下方点击 Add a user。 输入用户名和密码例如 用户名admin 密码admin123 点击 Add user 完成创建。
2. 分配权限 在用户列表中找到刚创建的用户如 admin。 点击用户右侧的 Set permission。 在权限设置页面 Virtual Host选择 /默认的虚拟主机。 Configure输入 .*表示允许用户配置所有资源。 Write输入 .*表示允许用户写入所有资源。 Read输入 .*表示允许用户读取所有资源。 点击 Set permission 完成权限分配。
3. 使用新用户登录 退出当前用户点击右上角的 guest选择 Log out。 使用新用户如 admin登录。
2.4 Spring Boot 中引入 RabbitMQ 依赖
在 pom.xml 中添加以下依赖
dependencies!-- RabbitMQ 依赖 --dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency
/dependencies
spring-boot-starter-amqp 是 Spring Boot 提供的 RabbitMQ 集成依赖它包含了以下内容 RabbitMQ 客户端库 自动引入 RabbitMQ 的 Java 客户端库amqp-client用于与 RabbitMQ 服务器通信。 Spring AMQP 支持 提供了 Spring 对 AMQPAdvanced Message Queuing Protocol的支持包括 RabbitTemplate、RabbitListener 等。 2.5 Spring Boot 配置 RabbitMQ
在 Spring Boot 项目中您需要在 application.properties 或 application.yml 中配置 RabbitMQ 的连接信息。
示例配置
# RabbitMQ 连接配置
spring.rabbitmq.hostlocalhost
spring.rabbitmq.port5672
spring.rabbitmq.usernameadmin
spring.rabbitmq.passwordadmin123
配置说明 spring.rabbitmq.hostRabbitMQ 服务器地址默认 localhost。 spring.rabbitmq.portRabbitMQ 消息通信端口默认 5672。 spring.rabbitmq.usernameRabbitMQ 用户名。 spring.rabbitmq.passwordRabbitMQ 密码。 3. Spring Boot 集成 RabbitMQ 的消息生产和消费
3.1 消息生产者Producer
在 Spring Boot 中我们使用 RabbitTemplate 来发送消息。它由 spring-boot-starter-amqp 自动配置成为一个 Bean可直接通过 Autowired 注入。
如果 message 不是 String 类型的处理
Spring AMQPspring-boot-starter-amqp在使用 RabbitTemplate 时默认的消息转换器MessageConverter通常会将对象序列化为 JSON 或者将字符串消息转换为字节。如果你的业务数据不是 String常见做法是 在发送时把非字符串对象序列化如转换为 JSON 字符串或者配置自定义的 MessageConverter让 Spring 帮你把对象自动序列化/反序列化。
典型做法手动序列化为 JSON 再发送
Service
public class CustomObjectProducer {Autowiredprivate RabbitTemplate rabbitTemplate;public void sendCustomObject(String queueName, MyCustomObject obj) {// 1. 将自定义对象序列化为 JSON 字符串String jsonString new Gson().toJson(obj);// 2. 发送 JSON 字符串到 RabbitMQrabbitTemplate.convertAndSend(queueName, jsonString);}
}在消费者端你也可以将消息JSON 字符串反序列化为 MyCustomObject。
配置自定义 Converter可选
Spring AMQP 提供了 Jackson2JsonMessageConverter 等现成转换器。
Configuration
public class RabbitConfig {Beanpublic MessageConverter jsonMessageConverter() {return new Jackson2JsonMessageConverter();}// 配置 RabbitTemplate 使用该转换器Beanpublic RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {RabbitTemplate template new RabbitTemplate(connectionFactory);template.setMessageConverter(jsonMessageConverter());return template;}
}这样一来rabbitTemplate.convertAndSend(queueName, myObject) 会自动把 myObject 转成 JSON 发送消费者端则自动解析为同样的 Java 对象。
1基本消息发送
场景 将消息直接发送到指定的队列跳过交换机的路由让 RabbitMQ 把消息放到这个队列中。
核心代码示例
Service
public class BasicProducer {Autowiredprivate RabbitTemplate rabbitTemplate; // 1.自动注入的 RabbitTemplate/*** 2.发送基本消息到指定的队列* param queueName 目标队列名称* param message 消息内容*/public void sendToQueue(String queueName, String message) {// 3.调用 convertAndSend直接将消息放入指定队列rabbitTemplate.convertAndSend(queueName, message);System.out.println(Message sent to queue: queueName , content: message);}
}代码详解 Autowired private RabbitTemplate rabbitTemplate; Spring Boot 自动为我们配置了 RabbitTemplate不用手动定义 Bean。 通过依赖注入即可使用所有与 RabbitMQ 交互的方法。 public void sendToQueue(String queueName, String message) 方法参数包括 queueName: 目标队列的名称。 message: 要发送的字符串类型消息内容。 rabbitTemplate.convertAndSend(queueName, message) convertAndSend 方法会将消息转换转换为字节并发送到指定队列。 如果该队列不存在RabbitMQ 会尝试自动创建前提是 Broker 端配置允许自动创建队列。 2发送到交换机
场景 将消息发送到一个交换机Exchange再由交换机通过 Routing Key 将消息路由到匹配的队列中。
核心代码示例
Service
public class ExchangeProducer {Autowiredprivate RabbitTemplate rabbitTemplate;/*** 发送消息到指定交换机* param exchangeName 交换机名称* param routingKey 路由键* param message 消息内容*/public void sendToExchange(String exchangeName, String routingKey, String message) {// 将消息发送到 exchangeName 指定的交换机使用 routingKey 进行路由rabbitTemplate.convertAndSend(exchangeName, routingKey, message);System.out.println(Message sent to exchange: exchangeName with routingKey: routingKey);}
}代码详解 exchangeName 要发送到的交换机名称例如 direct-exchange、fanout-exchange 等。 routingKey 路由键用来匹配绑定Binding。例如对 DirectExchange 而言需要队列绑定时的路由键与发送时的路由键相同消息才能到达队列。 rabbitTemplate.convertAndSend(exchangeName, routingKey, message) 将消息先发送到交换机再根据路由键将消息投递到目标队列。 3发送带消息属性的消息
场景 需要为消息设置 TTL过期时间或优先级等属性控制消息在队列中的行为。
核心代码示例
Service
public class PropertyProducer {Autowiredprivate RabbitTemplate rabbitTemplate;/*** 发送带消息属性的消息如 TTL, 优先级*/public void sendMessageWithProperties(String exchange, String routingKey, String messageContent) {// 1.创建 MessageProperties 对象用于指定消息的属性MessageProperties properties new MessageProperties();properties.setExpiration(10000); // 过期时间10秒 (单位毫秒)properties.setPriority(5); // 优先级设为 5// 2.根据消息体和属性构建 Message 对象Message message new Message(messageContent.getBytes(), properties);// 3.使用 send 方法(而非 convertAndSend)直接发送 Message 对象rabbitTemplate.send(exchange, routingKey, message);System.out.println(Message with properties sent: messageContent);}
}代码详解 MessageProperties properties new MessageProperties(); MessageProperties 用于设置 AMQP 协议层的各种消息头信息。 properties.setExpiration(10000); setExpiration 设置消息的 TTLTime-To-Live单位是毫秒。如果到达时间后消息仍未被消费RabbitMQ 会将其从队列中移除并送入死信队列如果配置了死信队列。 properties.setPriority(5); 设置消息的优先级为 5前提是队列本身需要支持优先级队列创建队列时指定 x-max-priority。 new Message(messageContent.getBytes(), properties) 将纯文本消息转换为 Message 对象结合了消息属性和消息体。 rabbitTemplate.send(exchange, routingKey, message); 与 convertAndSend 不同它不会尝试进行消息转换如 JSON、字符串而是直接发送完整的 AMQP Message 对象。 Message 构造函数 public Message(byte[] body, MessageProperties messageProperties) {this.body body;this.messageProperties messageProperties;
}body消息体的字节数组。messagePropertiesAMQP 的消息属性包括 TTL、优先级、headers 等。、 如果消息体不是String类型 手动转换为字节你可以先将自定义对象转换为字节数组例如通过 JSON 序列化或 Java 序列化再放入 new Message(...) 的第一个参数。 MyCustomObject obj new MyCustomObject();
// 假设你想用 JSON
String jsonString new Gson().toJson(obj);
byte[] body jsonString.getBytes(StandardCharsets.UTF_8);MessageProperties properties new MessageProperties();
// 设置一些属性
Message message new Message(body, properties);为什么不会自动转 JSON使用 new Message(...) 构造方法是“纯” AMQP 层的做法不会调用 Spring 的转换器因此你必须自己处理序列化。使用 Message 构造函数 时你必须自行处理对象到 byte[] 的转换无论是字符串、JSON还是其他格式。如果想让 Spring AMQP 自动转换你通常使用 rabbitTemplate.convertAndSend(Object msg) 这种高级 API或者配置自定义 MessageConverter。 3.2 消息消费者Consumer
消费者的核心功能是在指定的队列中监听消息并根据配置的确认模式自动确认或手动确认对消息进行处理或拒绝。
1监听队列并消费消息
核心代码示例自动确认模式
Service
public class Consumer {/*** 使用注解 RabbitListener 指定要监听的队列* 由于默认为 auto-ack 模式* 当消息到达后RabbitMQ 会自动确认并从队列中删除该消息。*/RabbitListener(queues demo-queue)public void receiveMessage(String message) {// 1.从 queueName 队列中取到的消息内容System.out.println(Received message: message);// 2.在 auto-ack 模式下无需手动 ack// 如果这里出现异常RabbitMQ 不会再次发送消息给消费者消息会丢失。}
}代码详解自动确认模式 RabbitListener(queues demo-queue) 声明监听名为 demo-queue 的队列。一旦有新消息到达该队列就会自动回调此方法。 public void receiveMessage(String message) 默认参数类型为字符串当 RabbitMQ 收到消息后会尝试将其转换为 String 并注入到 message 中。 自动确认auto-ack的风险 如果消费者在处理消息时抛出异常消息已经被 RabbitMQ 标记为“已确认”不会再重新发送或进入死信队列导致消息丢失。 2确认机制
自动确认auto-ack 行为 当消费者从队列中获取消息后RabbitMQ 会立即将该消息标记为已确认acknowledged并从队列中删除。 问题 如果消息处理失败例如消费者抛出异常消息已经被确认并从队列中删除无法重新处理。 如果消费者崩溃或断开连接未处理的消息会丢失。 适用场景 对消息处理的可靠性要求不高的场景。 手动确认manual-ack 行为 消费者处理完消息后必须显式调用 basicAck 方法确认消息。 如果消息处理失败可以调用 basicNack 或 basicReject 方法拒绝消息。 优点 确保消息处理的可靠性。 支持消息重新入队或发送到死信队列。 适用场景 对消息处理的可靠性要求较高的场景。
核心代码示例
Service
public class ManualAckConsumer {/*** 在 application.properties 中配置* spring.rabbitmq.listener.simple.acknowledge-modemanual* 使得 RabbitMQ 使用手动确认模式*/RabbitListener(queues demo-queue)public void receiveMessage(Message message, Channel channel) throws IOException {try {// 1.从消息中获取消息体String body new String(message.getBody());System.out.println(Processing message: body);// 2.如果业务处理成功则调用 basicAck 手动确认channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);} catch (Exception e) {System.err.println(Message processing failed: e.getMessage());// 3.如果处理失败需要决定是重新入队还是拒绝并进入死信队列// requeue true - 重新入队// requeue false - 丢弃或进入死信队列根据队列配置channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);}}
}代码详解 配置手动确认 在 application.properties 添加 spring.rabbitmq.listener.simple.acknowledge-modemanual表示 Spring AMQP 使用手动确认模式manual-ack。 public void receiveMessage(Message message, Channel channel) 与自动确认不同这里不仅接收字符串还接收了 org.springframework.amqp.core.Message 对象和 com.rabbitmq.client.Channel。Message包含消息体body和消息属性headers 等。Channel给我们提供了 basicAck, basicNack, basicReject 等底层 AMQP 操作。 手动确认成功 channel.basicAck(deliveryTag, multiple) deliveryTag本次消息的唯一标记从 message.getMessageProperties().getDeliveryTag() 获取。multiple false只确认当前这条消息。 basicAck(long deliveryTag, boolean multiple) 这里的 deliveryTag 并不是在你构造 Message 时生成的而是 RabbitMQ Broker 在投递消息给消费者时由底层 AMQP 协议自动分配的一个递增的序号。 long deliveryTag message.getMessageProperties().getDeliveryTag();手动确认失败 channel.basicNack(deliveryTag, multiple, requeue) 或 basicReject requeue true将消息重新放回队列等待下一次消费可能导致死循环如处理一直失败。requeue false拒绝消息若配置了死信队列则进入死信队列否则丢弃消息。 3处理消费失败 自动确认模式下的处理 在自动确认模式下如果消息处理失败RabbitMQ 不会重新发送消息因为消息已经被确认并从队列中删除。 问题 消息丢失无法重新处理。
手动确认模式下的处理
在手动确认模式下如果消息处理失败可以通过以下方式处理 重新入队 调用 basicNack 或 basicReject 方法并将 requeue 参数设置为 true。 消息会重新进入队列等待下一次消费。 发送到死信队列 调用 basicNack 或 basicReject 方法并将 requeue 参数设置为 false。 如果队列配置了死信队列消息会被发送到死信队列。
重试机制Spring AMQP 提供的简单重试只支持手动确认机制 是重试失败了才会将消息重新入队 所以重试在前重新入队在后 # 启用重试
spring.rabbitmq.listener.simple.retry.enabledtrue
# 最大重试次数
spring.rabbitmq.listener.simple.retry.max-attempts3
# 初始重试间隔
spring.rabbitmq.listener.simple.retry.initial-interval1000
# 间隔倍数
spring.rabbitmq.listener.simple.retry.multiplier2.0
# 最大重试间隔
spring.rabbitmq.listener.simple.retry.max-interval10000Spring AMQP 提供了 重试机制可以在消费者处理消息失败时自动进行多次重试而不是直接将消息重新入队。
行为 当消息处理失败时Spring AMQP 会在 本地 进行重试即不将消息重新入队直到达到最大重试次数。 如果重试次数用尽消息会被拒绝basicNack 或 basicReject并根据配置决定是否重新入队或发送到死信队列。
死信队列DLQ 当消息被拒绝或过期时RabbitMQ 会将其发送到我们配置的死信交换机DLX再路由到死信队列DLQ。 配置示例 Configuration
public class RabbitConfig {Beanpublic Queue normalQueue() {return QueueBuilder.durable(normal-queue).withArgument(x-dead-letter-exchange, dead-letter-exchange) // 指定死信交换机.withArgument(x-dead-letter-routing-key, dead-letter-routing-key) // 指定死信路由键.build();}Beanpublic DirectExchange deadLetterExchange() {return new DirectExchange(dead-letter-exchange);}Beanpublic Queue deadLetterQueue() {return new Queue(dead-letter-queue);}Beanpublic Binding deadLetterBinding(Queue deadLetterQueue, DirectExchange deadLetterExchange) {return BindingBuilder.bind(deadLetterQueue).to(deadLetterExchange).with(dead-letter-routing-key);}
}原理 正常队列通过 x-dead-letter-exchange 指定死信交换机一旦消息被拒绝requeuefalse或超时TTL 到期RabbitMQ 会把消息发送到 dead-letter-exchange。dead-letter-exchange 与 dead-letter-queue 进行绑定路由键 dead-letter-routing-key从而实现死信队列的存储。 重新入队 vs 发送到死信队列 重新入队channel.basicNack(deliveryTag, false, true) 适用于临时性错误比如数据库锁冲突、网络抖动等等待后续重新处理。发送到死信队列channel.basicNack(deliveryTag, false, false) 适用于永久性错误比如消息格式无法解析或业务逻辑指定不应再尝试。
