RabbitMQ-入门

RabbitMQ-入门

基本说明

AMQP协议（高消息队列协议：Advanced Message Queuing Protocol）：https://www.rabbitmq.com/tutorials/amqp-concepts.html

基础组件术语

• 生产者（Publisher）：发消息到某个交换机
• 消费者（Consumer）：从某个队列中取消息
• 交换机（Exchange）：负责把消息转发到对应的队列
• 队列（Queue）：存储消息的
• 路由（Route）：转发，就是怎么把消息从一个地方转到另一个地方（比如从生产者转发到某个队列）

安装RabbitMQ

Windows安装

Windows官网安装教程：https://www.rabbitmq.com/install-windows.html

安装Erlang 25.3.2 ：因为Rabbit MQ 依赖于erlang ，它的性能非常高

Erlang 下载：https://www.erlang.org/patches/otp-25.3.2

Windows 安装 Rabbit MQ 监控面板，执行下面的命令：

rabbitmq-plugins.bat enable rabbitmq_management

Rabbit MQ 端口：

1. 5672：程序链接的端口
2. 15672：WebUI

Linux安装

安装Docker

sudo yum install docker-ce

安装完Docker后，启动Docker服务：

sudo systemctl start docker

通过 docke r拉取 rabbitmq 镜像：

sudo docker pull rabbitmq:版本号
docker pull rabbitmq:3.9-management

请将 “版本号” 替换为您想要的RabbitMQ版本号。如果不指定版本号，则会拉取最新的稳定版本。

完成镜像拉取后，可以使用以下命令启动RabbitMQ容器：

sudo docker run -d --name myrabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:版本号
sudo docker run -d --name myrabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3.9-management

同样，请将"版本号"替换为您之前选择的RabbitMQ版本号。

容器启动后，可以使用以下命令检查容器是否正在运行：

sudo docker ps

如果能看到刚刚启动的RabbitMQ容器正在运行，那么恭喜您已成功在CentOS上使用Docker安装了RabbitMQ。

还在将服务器的1562、5672端口开放，

最后，通过浏览器访问 http:// 服务器IP地址:15672 来打开RabbitMQ的WebUI管理界面，默认的用户名和密码是"guest"。可以进行用户设置、队列管理等操作。

Linux下的RabbitMQ新增一个用户，分别执行以下的命令：

# 查看rabbitmq 运行的容器id
dcker ps
## 进入正在运行的 RabbitMQ 容器的 Shell。执行以下命令（将 <container_id> 替换为实际的容器 ID）：
docker exec -it <container_id> /bin/bash
# 添加新用户
rabbitmqctl add_user 用户名 密码
# 管理员权限
rabbitmqctl set_user_tags 上面的用户名 administrator 
# 列出用户列表
rabbitmqctl list_users
# 目录权限
# 第一个".*" 用于在每个实体上配置权限
# 第二个".*" 表示对每个实体的写入权限
# 第三个".*” 对每个实体的读取权限
rabbitmqctl set_permissions -p "/" "用户名" ".*" ".*" ".*"

快速入门

单向发送

一个生产者给一个队列发送消息，一个消费者从这个队列中获取到消息。1 对 1 关系

引入Rabbit MQ 组件依赖

<!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->
<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.16.0</version>
</dependency>

• Producer-Code:

/*** 单消费者** @author Lazy* @create 2024-12-07 23:34*/
public class SingleProducer {private final static String SINGLE_QUEUE_NAME = "hello";public static void main(String[] args) throws Exception {// 创建连接工厂ConnectionFactory factory = new ConnectionFactory();// 设置rabbitmq 对应的信息factory.setHost("localhost");
//        factory.setUsername("guest");
//        factory.setPassword("guest");try (Connection connection = factory.newConnection();// 建立连接，创建频道Channel channel = connection.createChannel()) {// 创建消息队列channel.queueDeclare(SINGLE_QUEUE_NAME, false, false, false, null);// 发送消息String message = "Hello World";channel.basicPublish("", SINGLE_QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8));System.out.println(" [x] Sent '" + message + "'");}}
}

代码说明：

Channal频道：可以理解为操作消息队列的clinet(比如 JDBCClient、redisClient)，提供了和消息队列server建立通信的传输方法（为了复用连接，提高传输效率）。程序通过channel 操作rabbitmq（收发消息）

Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,Map<String, Object> arguments) throws IOException;

参数说明

• queueName：消息队列名称（注意，同名称的消息队列，只能用同样的参数创建一次）
• durable：消息队列重启后，消息是否丢失，持久化处理
• exclusive：是否只允许当前这个创建消息队列的连接操作消息队列
• autoDelete：没有人用队列后，是否要删除队列
• arguments：用于设置队列的其他属性。可以通过传递一个键值对的Map来指定这些属性。

• Consumer-Code:

/*** @author Lazy* @Description: 单消费者* @create 2024-12-07 23:49*/
public class SingleConsumer {private final static String SINGLE_QUEUE_NAME = "hello";public static void main(String[] args) throws Exception {// 建立连接ConnectionFactory factory = new ConnectionFactory();// 设置连接信息factory.setHost("localhost");Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 创建队列channel.queueDeclare(SINGLE_QUEUE_NAME, false, false, false, null);System.out.println(" [*] Waiting for messages. To exit press CTRL+C");// 定义如何处理消息DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), StandardCharsets.UTF_8);System.out.println(" [x] Received '" + message + "'");};// 消费消息，会持续阻塞channel.basicConsume(SINGLE_QUEUE_NAME, true, deliverCallback, consumerTag -> {});}
}

多消费者（使用较少）

场景：多个机器同时取接受并处理任务（尤其是每个机器的处理能力有限）

一个生产者给一个队列发消息，多个消费者从一个队列取消息。1对多关系。

1）队列持久化

第二个参数：durable参数设置为true，服务器重启后队列不丢失

channel.queueDeclare(TAKS_QUEUE_NAME,true,false,false,false,null)

2）消息持久化

将basicPublish中的第二个参数指定为MessageProperties.PERSISTENT_TEXT_PLAIN ：

channel.basicPublish("", TASK_QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes("UTF-8"));

• “”：表示交换机的名称，这里为空字符串，表示使用默认的交换机。
• TASK_QUEUE_NAME：表示要发布消息的队列的名称。
• MessageProperties.PERSISTENT_TEXT_PLAIN：指定消息的属性，这里是设置消息持久化。PERSISTENT_TEXT_PLAIN是一个MessageProperties对象，它可以设置消息的各种属性，比如持久化、优先级等。
• message.getBytes(“UTF-8”)：将消息内容转换为字节数组，并指定字符集为UTF-8。

这个方法会将消息发送到指定的队列中，如果队列不存在则会抛出异常。消息的持久化属性设置为true，表示消息会被存储到磁盘上，即使在服务器重启后也能保留。

• Producer-Code:

/*** @author Lazy* @Description: 多生产者* @create 2024-12-08 0:27*/
public class MultiProducer {private static final String TASK_QUEUE_NAME = "multi_queue";public static void main(String[] args) 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);Scanner scanner = new Scanner(System.in);while (scanner.hasNext()) {String message = scanner.nextLine();// 发送消息channel.basicPublish("", TASK_QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + message + "'");}}}
}

• Consumer-Code:

/*** @author Lazy* @Description: 多消费者* @create 2024-12-08 0:27*/
public class MultiConsumer {private static final String TASK_QUEUE_NAME = "multi_queue";public static void main(String[] args) throws IOException, TimeoutException {// 建立连接ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");// 创建连接Connection connection = factory.newConnection();for (int i = 0; i < 2; i++) {// 创建频道final 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);// 定义如何处理接收到的消息int finalI = i;DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");try {// 处理工作System.out.println(" [x] Received '" + "编号:" + finalI + ":" + message + "'");channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);// 停 20 秒，模拟机器处理能力有限Thread.sleep(20000);} catch (InterruptedException e) {e.printStackTrace();channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false);} finally {System.out.println(" [x] Done");channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}};// 开启消费监听channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, consumerTag -> {});}}
}

控制单个消费者的处理任务积压数：每个消费者最多同时处理1个任务

channel.basicQos(1);

消费确认机制

channel.basicConsume(TASK_QUEUE_NAME,false,deliverCallback, consumerTag -> {});
// basicConsume()方法源码：
String basicConsume(String queue, boolean autoAck, DeliverCallback deliverCallback, CancelCallback cancelCallback) throws IOException;

为了保证消息成功被消费（快递成功被取走），rabbitmg提供了消息确认机制，当消费者接收到消息后，比如要给一个反馈：

• ack：消费成功
• nack：消费失败
• reject：拒绝

如果告诉RabbitMQ 服务器消费成功，服务器才会放心地移除消息

支持配置autoack，会自动执行ack命令，接收到消息立刻就成功了。

建议 autoack 改为 false ，根据实际情况，去手动确认。

指定确认某条消息

channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
// 源码
void basicAck(long deliveryTag, boolean multiple) throws IOException;

参数解释：

1. deliveryTag：表示要确认的消息的标识符。每个消息都有一个唯一的deliveryTage，用于标识消息顺序
2. multiple：表示是否批量确认消息。如果设置的true，则表示确认所有在deliveryTag之前的未确认消息；如果设置为false，则只确认当前deliveryTag的消息。

另外，第二个参数multiple批量确认：是指是否要一次性确认所有的历史消息直到当前这条

指定拒绝某条消息

channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);
// 源码
void basicAck(long deliveryTag, boolean multiple) throws IOException;

参数解释：

1. deliveryTag：表示要否定确认的消息的标识符。每个消息都有一个唯一的deliveryTag，用于标识消息的顺序。
2. multiple：表示是否批量否定确认消息。如果设置为true，则表示否定所有在deliveryTag之前的未确认消息；如果设置为false，则只否定当前deliveryTag的消息。
3. requeue：表示是否将消息重新放回队列。如果设置为true，则消息将被重新放回队列并可以被其他消费者重新消费；如果设置为false，则消息将会被丢弃。

第三个参数表示是否重新入队，可用于重试

2个测试小技巧：

1. 使用Scanner接受用户输入，便于快速发送多条消息
2. 使用 for 循环创建多个消费者，便于快速验证队列模型工作机制

交换机

一个生产者给多个队列发消息，1个生产者对多个队列。

交换机的作用：提供消息转发功能，类以于网络路由器

解决问题：怎么把消息转发到不同的队列上，好让消费者从不同的队列消费。

绑定：交换机和剃关联起来。

绑定代码方法：

channel.exchangeDeclare(FANOUT_EXCHANGE_NAME, "fanout");

后面的就是不同的队列

交换机类型

1. Fanout Exchange（广播类型）：Fanout交换机将消息广播到其绑定的所有队列。当消息被发送到Fanout交换机时，它会将消息复制到所有绑定的独立上，而不考虑路由键的值。因此，无论消息的路由键是什么，都会被广播到所有队列。Fanout交换机主要用于广播消息给所有的消费者。
2. Direct Exchange （直连类型）：Direct交换机是根据消息的路由键选择路由到与消息具有相同路由键绑定的队列。例如，当消息的路由键与绑定键完全匹配时，消息将被路由到对应的队列。Direct交换机主要用于一对一的消息路由。
3. Topic Exchange（主题类型）：Topic交换机将消息根据路由键的模式进行匹配，并将消息路由到与雄消息的路由键匹配的队列。路由键可以使用通配符匹配，支持两种通配符符号，"#“表示匹配一个或多个单词，”*"表示匹配一个单词。Topic交换机主要用于灵活的消息路由。
4. Headers Exchange（头类型）：Headers交换机是根据消息的头部信息进行匹配，并将消息路由到匹配的队列。头部信息通常是一组键值对，可以使用各种自定义的标准和非标准的头部信息进行匹配。Headers交换机主要用于复杂的匹配规则。

fanout交换机

扇出、广播消息

特点：消息会被转发到所有绑定到交换机的队列

场景：很适用于发布订阅的场景。比如写日志，可以多个系统间共享

实例场景：

• Producer-Code:

/*** @author Lazy* @Description: 广播交换机类型：fanout* @create 2024-12-08 16:48*/
public class FanoutProducer {private static final String EXCHANGE_NAME = "fanout_exchange";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {//声明fanout类型的交换机channel.exchangeDeclare(EXCHANGE_NAME, "fanout");Scanner scanner = new Scanner(System.in);while (scanner.hasNext()){String message = scanner.nextLine();channel.basicPublish(EXCHANGE_NAME,"nick", null, message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + PubSubOutput.Type.message + "'");}}}
}

• Consumer-Code:

1. 消费者和生产者要绑定同一个交换机
2. 先有队列，才能进行绑定

/*** @author Lazy* @Description: fanout 模式消费者* @create 2024-12-08 16:48*/
public class FanoutConsumer {private static final String EXCHANGE_NAME = "fanout_exchange";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");Connection connection = factory.newConnection();// 创建一个通道Channel channel = connection.createChannel();Channel channel2 = connection.createChannel();// 声明一个交换机channel.exchangeDeclare(EXCHANGE_NAME, "fanout");// 声明第一个队列String queueName = "lazy_queue01";channel.queueDeclare(queueName, true, false, false, null);channel.queueBind(queueName, EXCHANGE_NAME, "nick");// 声明第二个队列String queueName2 = "lazy_queue02";channel2.queueDeclare(queueName2, true, false, false, null);channel2.queueBind(queueName2, EXCHANGE_NAME, "nick");System.out.println(" [*] Waiting for messages. To exit press CTRL+C");DeliverCallback deliverCallback1 = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [lazy_queue01] Received '" + message + "'");};DeliverCallback deliverCallback2 = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [lazy_queue02] Received '" + message + "'");};channel.basicConsume(queueName, true, deliverCallback1, consumerTag -> {});channel.basicConsume(queueName2, true, deliverCallback2, consumerTag -> {});}
}

效果：所有的消费者都能收到消息

Direct交换机

官网介绍教程：https://www.rabbitmq.com/tutorials/tutorial-four-java.html

绑定：可以让交换机和队列进行关联，可以指定让交换机把什么样的消息发送给哪个队列（类似于计算机网络中，两个路由器，或者网络设备相互连接，也可以理解为网线）

routing Key：路由键，控制消息转发给指定的那个队列的（可以简单的说时IP地址）

特点：消息会根据路由键转发到指定的队列

场景：特定的消息只交给特定的系统（程序）来处理

绑定关系：完全匹配字符串，路由键要完全匹配

不同的队列也可以绑定同样的路由键

比如发日志的场景，希望用独立的程序来处理不同级别的日志，比如C1系统处理error日志，C2系统处理其他级别的日志。

示例场景

• Producer-Code：

/*** @author Lazy* @Description: 直连交换机类型direct——生产者* @create 2024-12-08 18:37*/
public class DirectProducer {private static final String DIRECT_EXCHANGER = "direct-exchange";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();// 创建一个通道Channel channel = connection.createChannel()) {// 声明一个交换机channel.exchangeDeclare(DIRECT_EXCHANGER, "direct");Scanner scanner = new Scanner(System.in);while (scanner.hasNext()) {// 获取用户输入String userInput = scanner.nextLine();// 判断用户输入是否合法String[] splits = userInput.split(" ");// 如果用户输入的命令不合法，则继续循环if (splits.length < 1) {continue;}String message = splits[0];String routingKey = splits[1];channel.basicPublish(DIRECT_EXCHANGER, routingKey, null, message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + message + " with routing " + routingKey + "'");}}}
}

• Consumer-Code:

/*** @author Lazy* @Description: 直连交换机类型direct——消费者* @create 2024-12-08 18:37*/
public class DirectConsumer {private static final String DIRECT_EXCHANGER = "direct-exchange";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");Connection connection = factory.newConnection();// 这里创建两个频道/*** 在使用消息队列（MQ）时，创建多个频道（Channel）来对应不同的队列（Queue）主要是为了实现不同的功能和需求。** 1. 一个频道对应一个队列：这种情况下，每个频道只负责与一个队列进行交互，可以实现对单个队列的独立操作，如发送消息、接收消息等。这种方式适用于需要对每个队列进行单独管理的场景。** 2. 一个频道对应两个队列：这种情况下，一个频道可以同时与两个队列进行交互，可以实现在两个队列之间进行消息的转发、订阅等操作。这种方式适用于需要在不同队列之间进行消息传递的场景。** 总的来说，创建一个频道对应一个队列还是一个频道对应两个队列，主要取决于你的业务需求和应用场景。如果需要对每个队列进行单独管理，可以使用一个频道对应一个队列的方式；如果需要在多个队列之间进行消息传递，可以使用一个频道对应多个队列的方式。*/// 第一个频道Channel channel1 = connection.createChannel();// 第二个频道Channel channel2 = connection.createChannel();// 输入模式： 消息: 路由键channel1.exchangeDeclare(DIRECT_EXCHANGER, "direct");String queueName = "lazy01_queue";channel1.queueDeclare(queueName, true, false, false, null);channel1.queueBind(queueName, DIRECT_EXCHANGER, "lazy01");// 输入模式： 消息: 路由键channel2.exchangeDeclare(DIRECT_EXCHANGER, "direct");String queueName2 = "lazy02_queue";channel2.queueDeclare(queueName2, true, false, false, null);//指定该队列绑定多个不同的路由键,该队列可以接收到多个路由键匹配发送来的消息String[] routingKeys = {"lazy01", "lazy02"};for (String routingKey : routingKeys) {channel2.queueBind(queueName2, DIRECT_EXCHANGER, routingKey);}System.out.println(" [*] Waiting for messages. To exit press CTRL+C");DeliverCallback deliverCallback1 = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [lazy01_queue] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");};DeliverCallback deliverCallback2 = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [lazy02_queue] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");};//消息队列监听channel1.basicConsume(queueName, true, deliverCallback1, consumerTag -> {});channel2.basicConsume(queueName2, true, deliverCallback2, consumerTag -> {});}
}

topic交换机

官网教程：https://www.rabbitmq.com/tutorials/tutorial-five-java.html

特点：消息会根据一个模糊的路由键转发到指定的队列

场景：特定的一类消息可以交给特定的一类系统（程序）来处理

绑定关系：可以模糊匹配多个绑定

1. *：匹配一个单词，比如*.lazy，那a.lazy、b.lazy都能匹配

2. #：匹配0个或多个单词，比如a.#，那么a.a、a.b、a.a.a都匹配

   #.a.# 可以匹配 a.b 、a1.a 等形式

   a的前面可以是0个或多个字符串，后面也是0个或者多个字符串

注意，这里的匹配和MySQL的like的%不一样，只按照单词来匹配，每个’.‘分隔单词，如果是’#.'，其实可以忽略，匹配0个词可以的

示例场景：

老板下发一个任务，让多个任务组都接受到这个任务消息

• Producer-Code:

/*** 主题交换机类型：Topic* @Description: 生产者* @author Lazy* @create 2024-12-08 21:32*/
public class TopicProducer {private static final String TOPIC_EXCHANGE = "topic-exchange";public static void main(String[] args) throws Exception{ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try(Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {channel.exchangeDeclare(TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC);Scanner scanner = new Scanner(System.in);while (scanner.hasNext()){String userInput = scanner.nextLine();String[] splits = userInput.split(" ");if(splits.length < 1){continue;}String message = splits[0];String routingKey = splits[1];channel.basicPublish(TOPIC_EXCHANGE, routingKey, null, message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + message + " with routing " + routingKey + "'");}}}
}

• Consumer-Code:

/*** 主题交换机类型：Topic** @author Lazy* @Description: 消费者* @create 2024-12-08 21:32*/
public class TopicConsumer {private static final String TOPIC_EXCHANGE = "topic-exchange";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");Connection connection = factory.newConnection();Channel channel = connection.createChannel();//输入模式：消息：路由键channel.exchangeDeclare(TOPIC_EXCHANGE, "topic");/*** 主题交换机Topic* 1、*：匹配一个单词* 2、#：匹配一个或多个单词** 发送消息——“项目B 前端.后端”* 使用“.前端.#”匹配路由键** 消费者接收：* 1、[后端B组] Received '前端.后端：项目B'* 2、[前端A组] Received '前端.后端：项目B'*/// 声明前端队列String queueName = "frontend_queue";channel.queueDeclare(queueName, true, false, false, null);channel.queueBind(queueName, TOPIC_EXCHANGE, "#.前端.#");// 声明后端队列String queueName2 = "backend_queue";channel.queueDeclare(queueName2, true, false, false, null);channel.queueBind(queueName2, TOPIC_EXCHANGE, "#.后端.#");// 声明产品队列String queueName3 = "product_queue";channel.queueDeclare(queueName3, true, false, false, null);channel.queueBind(queueName3, TOPIC_EXCHANGE, "#.产品.#");System.out.println(" [*] Waiting for messages. To exit press CTRL+C");DeliverCallback deliverCallback1 = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [前端A组] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");};DeliverCallback deliverCallback2 = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [后端B组] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");};DeliverCallback deliverCallback3 = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [产品C组] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");};//消息队列监听channel.basicConsume(queueName, true, deliverCallback1, consumerTag -> {});channel.basicConsume(queueName2, true, deliverCallback2, consumerTag -> {});channel.basicConsume(queueName3, true, deliverCallback3, consumerTag -> {});}
}

Headers交换机

类比主题和直接交换机，可以根据headers中的内容来指定发送到哪个队列。

由于性差、比较复杂，一般不推荐使用。

rabbitmq中的 headers 交换机是什么？有什么用？什么场景下会用？有什么优缺点？

AI答：

在RabbitMQ消息队列中，Headers交换机是一种特殊类型的交换机，它使用消息头（Headers）中的键值对来进行消息路由。与其他类型的交换机不同，Headers交换机无视消息的路由键，而是根据消息头中的键值对来决定消息的路由。

Headers交换机的主要用途是基于消息的附加属性来进行路由，而不是通过固定的路由键进行匹配。它适用于一些特殊的场景，例如需要根据消息的特定属性来进行过滤、路由或匹配的情况。Headers交换机可以让我们在消息发布的时候根据消息头的键值对来指定消息的目标队列或者路由规则。

优点：

• 灵活性高：Headers交换机可以根据消息的附加属性进行灵活的路由和匹配，可以满足一些特殊需求。
• 可以进行复杂的匹配：Headers交换机支持复杂的匹配规则，可以根据多个属性进行匹配和路由。

缺点：

• 性能相对较低：Headers交换机需要对所有消息的消息头进行匹配，因此在消息数量较大的情况下性能会有所下降。
• 配置复杂：Headers交换机需要明确指定消息头的键值对作为匹配条件，因此在配置时相对较为复杂。

在实际应用中，Headers交换机常常用于一些特定的场景，例如：

• 根据消息的自定义属性进行过滤和路由，以达到消息的分发和消费的目的。
• 对消息进行定制化的路由，根据消息头中的自定义属性选择不同的消费者进行处理。

需要注意的是，Headers交换机并不是常见的交换机类型，一般情况下，使用Direct、Fanout和Topic交换机就可以满足大多数的需求。只有在需要根据消息的附加属性进行路由或过滤时才考虑使用Headers交换机。

RPC：远程过程调用

支持用消息队列来模拟RPC的调用，但是一般没必要，直接用Dubbo、GRPC等RPC框架就好了。

实现一个场景总有更合适的、更专注的技术。一般都是直接使用Dubbo、GRPC等RPC框架技术即可。

在RabbitMQ消息队列中，RPC（Remote Procedure Call，远程过程调用）是一种通信模式，用于实现分布式系统中的远程调用。RPC允许一个进程（称为客户端）调用另一个进程（称为服务端）上的某个函数并获取返回结果，就像在本地调用函数一样。

RPC的使用可以简化分布式系统间的通信，提供更方便的远程调用方式。客户端可以将请求参数封装成消息发送到消息队列的一个特定队列中，服务端监听该队列并接收消息，然后执行相应的处理逻辑，并将处理结果发送回客户端。

使用RPC的主要目的是实现分布式系统的协同工作，例如：

• 将计算任务分发到不同的节点上进行并行处理，提高系统的性能和响应速度
• 实现微服务架构中的服务间的函数调用。

优点：

• 解耦性：RPC允许服务端和客户端通过异步消息传递进行通信，减少了服务之间的直接依赖。
• 可扩展性：RPC可以方便地加入新的服务或者移除不再需要的服务，通过消息队列可以实现动态的服务发现和注册。
• 并发处理：RPC可以实现并发处理多个请求，提高系统的吞吐量和并发能力。

在选择使用RPC时，需要考虑系统的性能需求和规模，确保消息队列的性能能够满足RPC的通信需求。同时，还需要合理设置RPC的超时和重试机制，以应对网络故障或服务不可用的情况。

核心特性

消息过期机制

官网：https://www.rabbitmq.com/ttl.html

可以给每条消息指定一个有效期，一段时间内未被消费者处理，就过期了。

示例场景：

• 消费者（库存系统）挂了，一个订单15分钟还没被库存系统处理，这个订单其实已经失效了，哪怕库存系统再恢复，其实也不用扣减库存。

适用场景：清理过期数据、模拟延迟队列的实现（不开会员就慢速）、专门让某个程序处理过期请求

1. 订单超时取消：在电商平台中，可以设置订单的过期时间，如果用户在规定时间内未支付订单，则将订单标记为过期取消。
2. 预约失效处理：在医院或美容院等场所，用户预约服务后，可以设置预约消息的过期时间，如果用户在规定时间内未到达，可以取消预约或释放时间。
3. 缓存更新和失效：在网站或应用中，可以将数据加载到缓存中，设置过期时间以保持数据的新鲜度，并在数据过期后重新加载最新数据。
4. 日志记录和清理：在系统中，可以将日志记录为消息发送到队列中，并设置过期时间以限制日志数据的保留时间，以及自动清理过期的日志。
5. 定时任务调度：可以使用RabbitMQ结合过期时间来实现定时任务调度。通过设置消息的过期时间，将任务消息发送到队列中，并在过期后触发执行相应的任务操作。

具体实现步骤

1）给队列中的所有消息指定过期时间

// 创建队列，指定消息过期参数
Map<String, Object> args = new HashMap<String, Object>();
args.put("x-message-ttl", 5000);
// args 指定参数
channel.queueDeclare(QUEUE_NAME, false, false, false, args);

如果在过期时间内，还没有消费者取消息，消息才会过期。

注意，如果消息已经接收到，但是没确认，是不会过期的。

ps：如果消息处于待消费状态并且过期时间到达后，消息将被标记为过期。但是，如果消息已经被消费者消费，并且正在处理过程中，即使过期时间到达，消息仍然会被正常处理。

• Producer-Code:

/*** @author Lazy* @Description: 消息过期生产者* @create 2024-12-08 22:40*/
public class TTLProducer {private final static String TTL_QUEUE = "ttl-queue";public static void main(String[] args) throws Exception {// 创建连接工厂ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel();) {// 创建消息队列 要删除掉，因为已经在消费者中创建了队列，没有必要再重新创建一次这个队列，如果在此处还创建队列，里面的参数必须要和消费者的参数一致// channel.queueDeclare(SINGLE_QUEUE_NAME, false, false, false, null);String message = "Hello World!";// 某条消息指定过期时间AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder().expiration("5000") // 设置消息的过期时间为5秒.build();channel.basicPublish("", TTL_QUEUE, null, message.getBytes(StandardCharsets.UTF_8));System.out.println(" [x] Sent '" + message + "'");}}
}

• Consumer-Code:

/*** @author Lazy* @Description: 消息超时消费者* @create 2024-12-08 22:40*/
public class TTLConsumer {private final static String TTL_QUEUE = "ttl-queue";public static void main(String[] args) throws Exception {// 建立链接，ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");Connection connection = factory.newConnection();// 创建通道Channel channel = connection.createChannel();// 声明队列HashMap<String, Object> msg = new HashMap<>();msg.put("x-message-ttl", 5000);// 指定args参数channel.queueDeclare(TTL_QUEUE, false, false, false, msg);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 + "'");};// 消费消息 autoAck设置为false 取消掉自动确认消息channel.basicConsume(TTL_QUEUE, false, deliverCallback, consumerTag -> {});}
}

2）给某条消息指定过期时间

语法：

// 给消息指定过期时间
AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder().expiration("1000").build();
channel.basicPublish("my-exchange", "routing-key", properties, message.getBytes(StandardCharsets.UTF_8));

消息确认机制

官方文档：https://www.rabbitmq.com/confirms.html

为了保证消息成功被消费（快递成功被取走），rabbitmq 提供了消息确认机制，当消费者接收到消息后，比如要给一个反馈：

• ack：消费成功
• nack：消费失败
• reject：拒绝

如果告诉 rabbitmq 服务器消费成功，服务器才会放心地移除消息。

支持配置 autoack，会自动执行 ack 命令，接收到消息立刻就成功了。

 channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, consumerTag -> {});

**一般情况，建议 autoack 改为 false，根据实际情况，去手动确认。** 
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), );

第二个参数 multiple 批量确认：是指是否要一次性确认所有的历史消息直到当前这条

指定拒绝某条消息：

channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false);

第 3 个参数表示是否重新入队，可用于重试

死信队列

官方文档：https://www.rabbitmq.com/dlx.html

为了保证消息的可靠性，比如每条消息都成功消费，需要提供一个容错机制，即：失败的消息怎么处理？

死信：过期的消息、拒收的消息、消息队列满了、处理失败的消息的统称

死信队列：专门处理死信的队列（注意，它就是一个普通队列，只不过是专门用来处理死信的，你甚至可以理解这个队列的名称叫 “死信队列”）

死信交换机：专门给死信队列转发消息的交换机（注意，它就是一个普通交换机，只不过是专门给死信队列发消息而已，理解为这个交换机的名称就叫 “死信交换机”）。也存在路由绑定

死信可以通过死信交换机绑定到死信队列。

示例场景：

实现：

1）创建死信交换机和死信队列，并且绑定关系

参考代码：

            // 声明死信交换机channel.exchangeDeclare(DEAD_EXCHANGE_NAME, "direct");// 加入boss队列String queueName = "boss_dlx_queue";channel.queueDeclare(queueName, true, false, false, null);channel.queueBind(queueName, DEAD_EXCHANGE_NAME, "boss");// 加入worker队列String queueName2 = "manage_dlx_queue";channel.queueDeclare(queueName2, true, false, false, null);channel.queueBind(queueName2, DEAD_EXCHANGE_NAME, "manage");

2）给失败之后需要容错处理的队列绑定死信交换机

示例代码：

//指定死信队列参数Map<String, Object> args01 = new HashMap<>();// 要绑定到哪个交换机args01.put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);// 指定死信要转发到哪个死信队列args01.put("x-dead-letter-routing-key", "boss");// 设置队列过期时间args01.put("x-message-ttl", 10000);String queueName = "dog_queue";channel.queueDeclare(queueName, true, false, false, args01);channel.queueBind(queueName, WORK_EXCHANGE_NAME, "dog");//指定死信队列参数Map<String, Object> args02 = new HashMap<>();// 要绑定到哪个交换机args02.put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);// 指定死信要转发到哪个死信队列args02.put("x-dead-letter-routing-key", "manage");String queueName2 = "cat_queue";channel.queueDeclare(queueName2, true, false, false, args02);channel.queueBind(queueName2, WORK_EXCHANGE_NAME, "cat");	

3）可以给要容错的队列指定死信之后的转发规则，死信应该再转发到哪个死信队列

// 指定死信要转发到哪个死信队列
args01.put("x-dead-letter-routing-key", "boss");

4）可以通过程序来读取死信队列中的消息，从而进行处理

• Producer-Code:

/*** 死信队列** @author Lazy* @create 2024-12-09 0:31*/
public class DlxdirectProducer {private static final String WORK_EXCHANGE_NAME = "direct2_exchange";private static final String DEAD_EXCHANGE_NAME = "dead_direct_exchange";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel();) {// 声明死信交换机channel.exchangeDeclare(DEAD_EXCHANGE_NAME, "direct");// 加入boss队列String queueName = "boss_dlx_queue";channel.queueDeclare(queueName, true, false, false, null);channel.queueBind(queueName, DEAD_EXCHANGE_NAME, "boss");// 加入worker队列String queueName2 = "manage_dlx_queue";channel.queueDeclare(queueName2, true, false, false, null);channel.queueBind(queueName2, DEAD_EXCHANGE_NAME, "manage");// 声明交换机DeliverCallback deliverCallback1 = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");// 拒绝消息channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false);System.out.println(" [boss] Received '" +delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");};DeliverCallback deliverCallback2 = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");// 拒绝消息channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false);System.out.println(" [manage] Received '" +delivery.getEnvelope().getRoutingKey() + "':'" + message + "'");};//消息队列监听channel.basicConsume(queueName, false, deliverCallback1, consumerTag -> {});channel.basicConsume(queueName2, false, deliverCallback2, consumerTag -> {});Scanner scanner = new Scanner(System.in);while (scanner.hasNext()) {String userInput = scanner.nextLine();String[] split = userInput.split(" ");if (split.length < 1) {continue;}String message = split[0];String routingKey = split[1];//设置消息过期时间，模拟死信channel.basicPublish(WORK_EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + message + "with routing:" + routingKey + "'");}}}
}

• Consumer-Code:

/*** 死信队列** @author Lazy* @create 2024-12-09 0:31*/
public class DlxdirectConsumer {private static final String WORK_EXCHANGE_NAME = "direct2_exchange";private static final String DEAD_EXCHANGE_NAME = "dead_direct_exchange";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.exchangeDeclare(WORK_EXCHANGE_NAME, BuiltinExchangeType.DIRECT);//指定死信队列参数Map<String, Object> args01 = new HashMap<>();// 要绑定到哪个交换机args01.put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);// 指定死信要转发到哪个死信队列args01.put("x-dead-letter-routing-key", "boss");// 设置队列过期时间args01.put("x-message-ttl", 10000);String queueName = "dog_queue";channel.queueDeclare(queueName, true, false, false, args01);channel.queueBind(queueName, WORK_EXCHANGE_NAME, "dog");//指定死信队列参数Map<String, Object> args02 = new HashMap<>();// 要绑定到哪个交换机args02.put("x-dead-letter-exchange", DEAD_EXCHANGE_NAME);// 指定死信要转发到哪个死信队列args02.put("x-dead-letter-routing-key", "manage");String queueName2 = "cat_queue";channel.queueDeclare(queueName2, true, false, false, args02);channel.queueBind(queueName2, WORK_EXCHANGE_NAME, "cat");System.out.println(" [*] Waiting for messages. To exit press CTRL+C");DeliverCallback deliverCallback1 = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");//拒绝消息channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false);System.out.println(" [dog] Received '" + delivery.getEnvelope().getRoutingKey() + "：" + message + "'");};DeliverCallback deliverCallback2 = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");//拒绝消息channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false);System.out.println(" [cat] Received '" + delivery.getEnvelope().getRoutingKey() + "：" + message + "'");};//消息队列监听channel.basicConsume(queueName, false, deliverCallback1, consumerTag -> {});channel.basicConsume(queueName2, false, deliverCallback2, consumerTag -> {});}
}

延迟队列

官网地址：https://blog.rabbitmq.com/posts/2015/04/scheduling-messages-with-rabbitmq

延迟队列（Delayed Queue），它允许消息在一定的延迟时间后被消费

作用：在消息到达队列后，不立即将消息投递给消费者，而是在一定延迟时间后再进行投递。延迟队列通常用于需要延迟处理的业务场景。延迟队列主要用于处理需要在特定时间后执行的任务或延迟消息。它可以为消息设置一个延迟时间，在指定的延迟时间后，消息会被自动投递到指定的消费者。例如定时任务，消息重试，延迟通知等。

延迟队列适用于许多场景，包括：

1. 定时任务：可以使用延迟队列来实现任务的定时触发，例如定时发送邮件或推送通知。
2. 消息重试：当某个消息失败后，可以将其放入延迟队列，并设置延迟时间，以便稍后重新投递
3. 延迟通知：例如在某个时间后发送提醒通知。

优点：

1. 灵活性：可以根据实际需求，灵活地设置延迟时间，适应各种业务场景。
2. 解耦性：延迟队列可以将消息发送与消费解耦，提高系统的可扩展性和稳定性。
3. 可靠性：通过延迟队列，可以确保消息在一定时间后被投递，降低消息丢失的风险。

缺点：

1. 系统复杂性：引入延迟队列会增加系统的复杂性和维护成本。
2. 延迟时间不准确：由于网络延迟、系统负载等原因，延迟时间可能会有一定的误差。

示例代码：

• Producer-Code：

/*** 延迟队列生产者** @author Lazy* @create 2024-12-10 17:56*/
public class DelayProducer {private static final String EXCHANGE_NAME = "order_exchange";private static final String DEL_EXCHANGE_NAME = "order_exchange_dlx";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel();) {// 声明交换机channel.exchangeDeclare(EXCHANGE_NAME, "direct");channel.exchangeDeclare(DEL_EXCHANGE_NAME, "direct");Scanner scanner = new Scanner(System.in);while (scanner.hasNext()) {String userInput = scanner.nextLine();String[] split = userInput.split(" ");if (split.length < 1) {continue;}String message = split[0];String routingKey = split[1];channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + message + "with routing:" + routingKey + "'");}}}
}

• Consumer-Code：

/*** 延迟队列消费者** @author Lazy* @create 2024-12-10 17:56*/
public class DelayConsumer {private static final String EXCHANGE_NAME = "order_exchange";private static final String DEL_EXCHANGE_NAME = "order_exchange_dlx";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");Connection connection = factory.newConnection();Channel channel = connection.createChannel();//Map<String, Object> args = new HashMap<>();args.put("x-dead-letter-exchange", DEL_EXCHANGE_NAME);args.put("x-dead-letter-routing-key", "orderdlx");args.put("x-message-ttl", 10000);//创建正常队列String queueName = "order_queue";channel.queueDeclare(queueName, true, false, false, args);channel.queueBind(queueName, EXCHANGE_NAME, "order");// 创建死信队列String queueName2 = "order_queue_dlx";channel.queueDeclare(queueName2, true, false, false, null);channel.queueBind(queueName2, DEL_EXCHANGE_NAME, "orderdlx");System.out.println(" [*] Waiting for messages. To exit press CTRL+C");DeliverCallback deliverCallback1 = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");//拒绝消息channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, false);System.out.println(" [order_queue] Received '" + delivery.getEnvelope().getRoutingKey() + "：" + message + "'");};DeliverCallback deliverCallback2 = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");//拒绝消息channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);System.out.println(" [order_queue_dlx] Received '" + delivery.getEnvelope().getRoutingKey() + "：" + message + "'");};//消息队列监听channel.basicConsume(queueName, false, deliverCallback1, consumerTag -> {});channel.basicConsume(queueName2, false, deliverCallback2, consumerTag -> {});}
}