【RabbitMQ】 相关概念 + 工作模式

本文将介绍一些MQ中常见的概念，同时也会简单实现一下RabbitMQ的工作流程。

MQ概念

Message Queue消息队列。是用来存储消息的队列，多用于分布式系统之间的通信。

系统间调用通常有：同步通信和异步通信。MQ就是在异步通信的时候使用的。

同步通信：

异步通信：

MQ作用

• 异步解耦：在业务流程中，一些操作可能非常耗时，但并不需要即时返回结果。可以借助消息队列（MQ）将这些操作异步化。例如，当用户注册后，可以将发送注册短信或邮件通知作为异步任务处理，而不必等待这些操作完成后才告知用户注册成功。
• 流量削峰：在访问量剧增的情况下，应用仍然需要继续发挥作用，但这样的突发流量并不常见。如果以能处理这类峰值为标准而投入资源，必然是巨大的浪费。使用MQ可以使关键组件支撑突发访问压力，不会因为突发流量而崩溃。例如，秒杀或促销活动中，可以使用MQ来控制流量，将请求排队，系统根据自己的处理能力逐步处理这些请求。
• 消息分发：当多个系统需要对同一数据做出响应时，可以使用MQ进行消息分发。例如，支付成功后，支付系统可以向MQ发送消息，其他系统订阅该消息，而无需轮询数据库。
• 延迟通知：在需要在特定时间后发送通知的场景中，可以使用MQ的延迟消息功能。例如，在电子商务平台中，如果用户下单后一定时间内未支付，可以使用延迟队列在超时后自动取消订单。

---

AMQP

AMQP（Advanced Message Queuing Protocol）是一个高级消息队列协议，定义了一系列消息交换功能，包括交换器（Exchange）和队列（Queue）。这些组件共同工作，使得生产者可以将消息发送到交换器，由队列接收并等待消费者取走。AMQP还定义了一个网络协议，允许客户端应用通过该协议与消息代理和AMQP模型进行交互通信。RabbitMQ遵循AMQP协议，使用Erlang实现，并支持其他协议如STOMP和MQTT。可以说AMQP模型和RabbitMQ模型结构一致。

---

RabbitMQ

功能完备: MQ 功能较为全面，几乎支持所有主流语言。
开源友好: 界面非常友好，易于使用。
性能: 性能良好，吞吐量可达到万级。
社区活跃度: 社区活跃度较高。
适用场景: 适合中小型公司，数据量和并发量较小的场景

---

Producer和Consumer

Producer（生产者）：是RabbitMQ Server的客户端，向RabbitMQ发送消息。消息通常带有业务逻辑结构的数据，例如JSON字符串，并可能包含标签，RabbitMQ根据标签路由消息。

Consumer（消费者）：也是RabbitMQ Server的客户端，从RabbitMQ接收消息。在消费过程中，标签会被丢弃，消费者只收到消息，且无需知道消息的生产者。

Broker（消息代理）：即RabbitMQ Server，负责接收和收发消息。通常，一个RabbitMQ Broker可以视为一个RabbitMQ服务节点或实例，大多数情况下等同于一台RabbitMQ服务器。

---

Connection和Channel

Connection（连接）：是客户端与RabbitMQ服务器之间的TCP连接。这是建立消息传递的基础，负责传输客户端和服务器之间的所有数据和控制信息。

Channel（通道/信道）：是在Connection之上的一个抽象层。在RabbitMQ中，一个TCP连接可以有多个Channel，每个Channel都是独立的虚拟连接。消息的发送和接收都是基于Channel进行的。通道的主要作用是将消息的读写操作复用到同一个TCP连接上，从而减少建立和关闭连接的开销，提高性能。

---

Virtual host

Virtual Host（虚拟主机）：这是一个虚拟概念，为消息队列提供逻辑上的隔离机制。在RabbitMQ中，一个Broker Server上可以存在多个Virtual Host。它允许不同的用户在同一RabbitMQ服务器上创建和管理独立的交换机（exchange）、队列（queue）等，类似于MySQL中的数据库（database）。

---

Queue

Queue: 队列, 是RabbitMQ的内部对象, 用于存储消息。

---

Exchange

Exchange（交换机）：是消息到达RabbitMQ Broker的第一站。它负责接收生产者发送的消息，并根据特定的规则将这些消息路由到一个或多个队列（Queue）中。Exchange的主要作用是消息路由，它根据类型和规则决定如何转发接收到的消息。

交换机类型

RabbitMQ中的交换机有四种主要类型，每种类型有不同的路由策略：

1. Fanout：将消息广播到所有绑定到该交换机的队列，无视路由键。
2. Direct：根据消息的路由键将消息精确路由到与交换机绑定的队列。
3. Topic：使用通配符模式匹配路由键，将消息路由到匹配的队列。
4. Headers：基于消息头部的属性进行路由，路由键不会被使用。（很少使用）

AMQP协议中还有另外两种交换机类型：

System：用于内部系统交换和管理，不用于常规的消息路由。
自定义（Custom）：用户可以定义自己的交换机类型，以满足特定的需求。

根据交换机类型的不同，创建了许多不同的工作模式。

Routing Key（路由键）
定义: 路由键是一个字符串，通过它生产者在将消息发送到交换器时告诉交换器如何处理该消息。
作用: 决定了消息的路由路径，即消息应发送到哪个队列或哪些队列。
Binding Key（绑定键）
定义: 在 RabbitMQ 中，通过 Binding（绑定）将交换器与队列关联时，通常会指定一个 Binding Key。
作用: 引导 RabbitMQ 知道如何将消息路由到相应的队列。这个键通常与路由键相关联，从而实现消息的过滤和分发。

通常来说路由键和绑定键没什么区别

---

工作流程

Producer（生产者） 产生了一条消息。
Producer 连接到 RabbitMQ Broker，建立一个连接（Connection），并开启一个信道（Channel）。
Producer 声明一个交换机（Exchange），用于路由消息。
Producer 声明一个队列（Queue），用于存放消息。
Producer 发送消息至 RabbitMQ Broker。
RabbitMQ Broker 接收消息，并将其存入相应的队列（Queue）中。如果未找到相应的队列，则根据生产者的配置，选择丢弃或退回消息给生产者。

---

工作模式

RabbitMQ提供了七个工作模式。查看文档即可：RabbitMQ Tutorials | RabbitMQ

工作模式图中的字母：

P（生产者）：也就是要发送消息的程序。

C（消费者）：消费者，消息的接收者。

Queue（消息队列）：缓存消息；生产者将消息投递到其中，消费者从中取出消息。

Exchange（交换机X）：按照规则分发消息。

相关通用代码提取到Constants类中

public class Constants {// 主机ippublic static final String HOST = "ip地址";// rabbitmq服务器端口号 默认是5672public static final int PORT = 5672;// rabbitmq用户名public static final String USER_NAME = "用户名";// rabbitmq用户密码public static final String PASSWORD = "用户密码";// rabbitmq虚拟主机public static final String VIRTUAL_HOST = "虚拟主机名";// 简单模式public static final String SIMPLE_QUEUE = "simple.queue";// 工作队列模式public static final String WORK_QUEUE = "work.queue";// 发布订阅模式public static final String FANOUT_EXCHANGE = "fanout.exchange";public static final String FANOUT_QUEUE1 = "fanout.queue1";public static final String FANOUT_QUEUE2 = "fanout.queue2";// 路由模式public static final String DIRECT_EXCHANGE = "direct.exchange";public static final String DIRECT_QUEUE1 = "direct.queue1";public static final String DIRECT_QUEUE2 = "direct.queue2";// 通配符模式public static final String TOPIC_EXCHANGE = "topic.exchange";public static final String TOPIC_QUEUE1 = "topic_queue1";public static final String TOPIC_QUEUE2 = "topic_queue2";// rpc 模式public static final String RPC_REQUEST_QUEUE = "rpc.request.queue";public static final String RPC_RESPONSE_QUEUE = "rpc.response.queue";// publisher confirmspublic static final String PUBLISHER_CONFIRMS_QUEUE1 = "publisher.confirms.queue1";public static final String PUBLISHER_CONFIRMS_QUEUE2 = "publisher.confirms.queue2";public static final String PUBLISHER_CONFIRMS_QUEUE3 = "publisher.confirms.queue3";// 推拉模式public static final String MESSAGE_QUEUE = "message.queue";
}

简单模式

特点: 一个生产者 P 和一个消费者 C，消息只能被消费一次。这种模式也称为点对点（Point-to-Point）模式。

适用场景: 当消息只能被单个消费者处理时。

---

生产者代码

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import rabbitmq.common.Constants;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Producer {public static void main(String[] args) throws IOException, TimeoutException {// 1. 创建链接ConnectionFactory factory = new ConnectionFactory();factory.setHost(Constants.HOST);factory.setPort(Constants.PORT);factory.setUsername(Constants.USER_NAME);factory.setPassword(Constants.PASSWORD);factory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection = factory.newConnection();// 2. 开启信道Channel channel = connection.createChannel();// 3. 声明交换机 简单模式使用内置交换机 所以不用声明// 4. 声明队列/** queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,*                                  Map<String, Object> arguments)*  参数说明:*  queue: 队列名称*  durable: 可持久化*  exclusive: 是否独占*  autoDelete: 是否自动删除*  arguments: 参数*/channel.queueDeclare(Constants.SIMPLE_QUEUE, true, false, false, null);// 5. 发送消息/** basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)* 参数说明:* exchange: 交换机名称 在简单模式下使用 "" 表示使用默认交换机* routingKey: 路由键 在简单模式下routingkey和队列名称保持一致* props: 属性配置* body: 消息*/for (int i = 0; i < 10; i++) {String msg = "hello rabbitmq simple " + i;channel.basicPublish("", Constants.SIMPLE_QUEUE, null, msg.getBytes());}System.out.println("消息发送成功");// 6. 关闭资源channel.close();connection.close();}
}

执行代码, 观察结果

---

消费者代码

import com.rabbitmq.client.*;
import rabbitmq.common.Constants;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer {public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {//1. 创建连接ConnectionFactory factory = new ConnectionFactory();factory.setHost(Constants.HOST);factory.setPort(Constants.PORT);factory.setUsername(Constants.USER_NAME);factory.setPassword(Constants.PASSWORD);factory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection = factory.newConnection();//2. 创建ChannelChannel channel = connection.createChannel();//3. 声明队列 如果没有该队列，会自动创建channel.queueDeclare(Constants.SIMPLE_QUEUE,true, false, false, null);//4. 消费消息// DefaultConsumer: 默认消费者DefaultConsumer consumer = new DefaultConsumer(channel){//从队列中收到消息, 就会执行的方法// 参数说明:// 1. 消息的唯一标识// 2. 消息的元信息// 3. 消息的属性// 4. 消息的内容@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接收到消息:"+ new String(body));}};/** basicConsume(String queue, boolean autoAck, Consumer callback)* 参数说明:* queue: 队列名称* autoAck: 是否自动确认* callback: 接收到消息后, 执行的逻辑*/channel.basicConsume(Constants.SIMPLE_QUEUE, true, consumer);//等待程序执行完成Thread.sleep(2000);//5. 释放资源channel.close();connection.close();}
}

执行代码, 观察结果

---

工作模式

每个消息会被分派给不同的消费者。消息不会重复。

适用场景:集群环境中进行异步处理。
例如：12306 短信通知服务。订单消息发送到 RabbitMQ，短信服务从 RabbitMQ 获取订单信息并发送通知，任务在短信服务之间进行分配。

---

生产者代码

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import rabbitmq.common.Constants;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Producer {public static void main(String[] args) throws IOException, TimeoutException {//1. 建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT); //需要提前开放端口号connectionFactory.setUsername(Constants.USER_NAME);//账号connectionFactory.setPassword(Constants.PASSWORD);  //密码connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST); //虚拟主机Connection connection = connectionFactory.newConnection();//2. 开启信道Channel channel = connection.createChannel();//3. 声明队列   使用内置的交换机//如果队列不存在, 则创建, 如果队列存在, 则不创建channel.queueDeclare(Constants.WORK_QUEUE, true, false, false, null);//4. 发送消息for (int i = 0; i < 10; i++) {String msg = "hello work queue " + i;channel.basicPublish("", Constants.WORK_QUEUE, null, msg.getBytes());}System.out.println("消息发送成功");//6. 资源释放channel.close();connection.close();}
}

执行代码, 观察结果

---

消费者代码

这里创建两个消费者

import com.rabbitmq.client.*;
import rabbitmq.common.Constants;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {//1. 建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT);connectionFactory.setUsername(Constants.USER_NAME);connectionFactory.setPassword(Constants.PASSWORD);connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();//2. 开启信道Channel channel = connection.createChannel();//3. 声明队列   使用内置的交换机//如果队列不存在, 则创建, 如果队列存在, 则不创建channel.queueDeclare(Constants.WORK_QUEUE, true, false, false, null);//4. 消费消息DefaultConsumer consumer = new DefaultConsumer(channel){//从队列中收到消息, 就会执行的方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接收到消息:"+ new String(body));}};channel.basicConsume(Constants.WORK_QUEUE, true, consumer);//6. 资源释放
//        channel.close();
//        connection.close();}
}

执行代码, 观察结果(这里让两个消费者跑起来,不要断开连接,然后再让生产者生产)

---

发布/订阅模式

工作流程:

发布者将消息发送到交换器，并指定一个主题。
交换器根据消息的主题将其路由到一个或多个队列。
订阅者从这些队列中获取消息。

特点:

广播: 消息可以被发送到多个订阅者。
解耦: 发布者和订阅者之间没有直接的联系，增加了系统的灵活性和可扩展性。

适用场景:

实时通知系统、事件驱动架构等，其中多个组件需要接收和处理相同的消息或事件。

---

生产者代码

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import rabbitmq.common.Constants;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Producer {public static void main(String[] args) throws IOException, TimeoutException {//1. 建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT);connectionFactory.setUsername(Constants.USER_NAME);connectionFactory.setPassword(Constants.PASSWORD);connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();//2. 开启信道Channel channel = connection.createChannel();//3. 声明交换机channel.exchangeDeclare(Constants.FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT, true);//4. 声明队列channel.queueDeclare(Constants.FANOUT_QUEUE1,true,false,false,null);channel.queueDeclare(Constants.FANOUT_QUEUE2,true,false,false,null);//5. 交换机和队列绑定//                  队列名                 交换机名                 路由键// 在 fanout的交换器中，路由键会被忽略。无论路由键是什么，交换器都会将消息广播到所有绑定的队列。因此，传递空字符串在这种情况下是合法的和无影响的。channel.queueBind(Constants.FANOUT_QUEUE1,Constants.FANOUT_EXCHANGE,"");channel.queueBind(Constants.FANOUT_QUEUE2,Constants.FANOUT_EXCHANGE,"");//6. 发布消息String msg = "hello fanout";channel.basicPublish(Constants.FANOUT_EXCHANGE,"", null, msg.getBytes());System.out.println("消息发送成功");//7. 释放资源channel.close();connection.close();}
}

---

消费者代码

import com.rabbitmq.client.*;
import rabbitmq.common.Constants;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {//1. 建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT);connectionFactory.setUsername(Constants.USER_NAME);connectionFactory.setPassword(Constants.PASSWORD);connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();//2. 开启信道Channel channel = connection.createChannel();//3. 声明队列channel.queueDeclare(Constants.FANOUT_QUEUE1,true,false,false,null);//4. 消费消息DefaultConsumer consumer = new DefaultConsumer(channel){//从队列中收到消息, 就会执行的方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接收到消息:"+ new String(body));}};channel.basicConsume(Constants.FANOUT_QUEUE1, true, consumer);//        // 5. 关闭资源
//        channel.close();
//        connection.close();}
}

这里两个消费者获取消息.

---

路由模式

描述: 路由模式是在发布/订阅模式的基础上增加了路由键（Routing Key）。在这种模式下，交换器（Exchange）根据消息的路由键规则，将消息筛选后发送到符合条件的消费者队列。

关键点:
交换器类型: 常使用 Direct 交换器。
路由键: 消息的路由键决定消息的去向。
绑定键: 队列与交换器的绑定时使用的键，用于确定队列接收哪些消息。

适用场景: 需要根据特定规则分发消息的场景。例如：
日志系统: 系统日志的等级包括 error、warning、info 和 debug。可以通过路由模式将不同等级的日志消息发送到不同的队列，最终输出到不同的文件。

---

生产者代码

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import rabbitmq.common.Constants;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** 路由模式生产者*/
public class Producer {public static void main(String[] args) throws IOException, TimeoutException {//1. 建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT);connectionFactory.setUsername(Constants.USER_NAME);connectionFactory.setPassword(Constants.PASSWORD);connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();//2. 开启信道Channel channel = connection.createChannel();//3. 声明交换机channel.exchangeDeclare(Constants.DIRECT_EXCHANGE, BuiltinExchangeType.DIRECT, true);//4. 声明队列channel.queueDeclare(Constants.DIRECT_QUEUE1, true, false, false, null);channel.queueDeclare(Constants.DIRECT_QUEUE2, true, false, false, null);//5. 绑定交换机和队列 绑定路由键// 队列1绑定路由键a// 队列2绑定路由键a b cchannel.queueBind(Constants.DIRECT_QUEUE1, Constants.DIRECT_EXCHANGE, "a");channel.queueBind(Constants.DIRECT_QUEUE2, Constants.DIRECT_EXCHANGE, "a");channel.queueBind(Constants.DIRECT_QUEUE2, Constants.DIRECT_EXCHANGE, "b");channel.queueBind(Constants.DIRECT_QUEUE2, Constants.DIRECT_EXCHANGE, "c");//6. 发送消息String msgA = "hello direct, my routingkey is a...";channel.basicPublish(Constants.DIRECT_EXCHANGE,"a", null, msgA.getBytes());String msgB = "hello direct, my routingkey is b...";channel.basicPublish(Constants.DIRECT_EXCHANGE,"b", null, msgB.getBytes());String msgC = "hello direct, my routingkey is c...";channel.basicPublish(Constants.DIRECT_EXCHANGE,"c", null, msgC.getBytes());System.out.println("消息发送成功");//7. 释放资源channel.close();connection.close();}
}

消费者代码

import com.rabbitmq.client.*;
import rabbitmq.common.Constants;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {//1. 建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT);connectionFactory.setUsername(Constants.USER_NAME);connectionFactory.setPassword(Constants.PASSWORD);connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();//2. 开启信道Channel channel = connection.createChannel();//3. 声明队列channel.queueDeclare(Constants.DIRECT_QUEUE1,true,false,false,null);//4. 消费消息DefaultConsumer consumer = new DefaultConsumer(channel){//从队列中收到消息, 就会执行的方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接收到消息:"+ new String(body));}};channel.basicConsume(Constants.DIRECT_QUEUE1, true, consumer);}
}

---

通配符模式

描述: Topic 模式是对路由模式的升级，增加了通配符功能，使消息路由更加灵活。生产者将消息发送给交换机，交换机根据路由键（Routing Key）将消息转发到匹配的队列。Topic 模式允许使用类似正则表达式的方式来定义路由键的模式，从而实现更复杂的匹配规则。

关键点:

交换器类型: 使用 Topic 交换器。
路由键模式: 支持通配符匹配，使路由更加灵活。通配符可以用来匹配多个值或匹配特定模式。

通配符:
*：匹配一个单词。
#：匹配零个或多个单词。

---

生产者代码

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import rabbitmq.common.Constants;import java.io.IOException;
import java.util.concurrent.TimeoutException;/*** 通配符模式生产者*/
public class Producer {public static void main(String[] args) throws IOException, TimeoutException {//1. 建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT);connectionFactory.setUsername(Constants.USER_NAME);connectionFactory.setPassword(Constants.PASSWORD);connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();//2. 开启信道Channel channel = connection.createChannel();//3. 声明交换机channel.exchangeDeclare(Constants.TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC, true);//4. 声明队列channel.queueDeclare(Constants.TOPIC_QUEUE1, true, false, false, null);channel.queueDeclare(Constants.TOPIC_QUEUE2, true, false, false, null);//5. 绑定交换机和队列// 队列1绑定 *.a.*// 队列2绑定 *.*.b 和 c.#channel.queueBind(Constants.TOPIC_QUEUE1, Constants.TOPIC_EXCHANGE, "*.a.*");channel.queueBind(Constants.TOPIC_QUEUE2, Constants.TOPIC_EXCHANGE, "*.*.b");channel.queueBind(Constants.TOPIC_QUEUE2, Constants.TOPIC_EXCHANGE, "c.#");//6. 发送消息String msg = "hello topic, my routingkey is ae.a.f....";channel.basicPublish(Constants.TOPIC_EXCHANGE,"ae.a.f", null, msg.getBytes());  //转发到Q1String msg_b = "hello topic, my routingkey is ef.a.b....";channel.basicPublish(Constants.TOPIC_EXCHANGE,"ef.a.b", null, msg_b.getBytes()); //转发到Q1和Q2String msg_c = "hello topic, my routingkey is c.ef.d....";channel.basicPublish(Constants.TOPIC_EXCHANGE,"c.ef.d", null, msg_c.getBytes());//转发Q2System.out.println("消息发送成功");//7. 释放资源channel.close();connection.close();}
}

消费者代码

import com.rabbitmq.client.*;
import rabbitmq.common.Constants;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {//1. 建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT);connectionFactory.setUsername(Constants.USER_NAME);connectionFactory.setPassword(Constants.PASSWORD);connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();//2. 开启信道Channel channel = connection.createChannel();//3. 声明队列channel.queueDeclare(Constants.TOPIC_QUEUE1,true,false,false,null);//4. 消费消息DefaultConsumer consumer = new DefaultConsumer(channel){//从队列中收到消息, 就会执行的方法@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("接收到消息:"+ new String(body));}};channel.basicConsume(Constants.TOPIC_QUEUE1, true, consumer);}
}

---

RPC通信

1. 客户端发送请求:客户端将消息发送到一个指定的请求队列。在消息属性中设置 replyTo 字段，指定一个回调队列用于接收服务端的响应。
2. 服务端处理请求:服务端从请求队列中接收并处理请求消息。服务端处理完成后，将响应消息发送到 replyTo 字段指定的回调队列。
3. 客户端接收响应:客户端在回调队列上等待响应消息。一旦收到响应，客户端会检查消息的 correlationId 属性，以确保该响应是与其请求相对应的。 

服务器代码

import com.rabbitmq.client.*;
import rabbitmq.common.Constants;import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;/*** RPC server* 1. 接收请求* 2. 发送响应*/
public class RpcServer {public static void main(String[] args) throws IOException, TimeoutException {//1. 建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT);connectionFactory.setUsername(Constants.USER_NAME);connectionFactory.setPassword(Constants.PASSWORD);connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();//2. 开启信道Channel channel = connection.createChannel();// 用来接收请求的队列channel.queueDeclare(Constants.RPC_REQUEST_QUEUE, true, false, false, null);// 用来发送响应的队列channel.queueDeclare(Constants.RPC_RESPONSE_QUEUE, true, false, false, null);//3. 接收请求channel.basicQos(1);DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String request = new String(body, StandardCharsets.UTF_8);System.out.println("接收到请求:"+ request);String response = "针对request:"+ request +", 响应成功";AMQP.BasicProperties basicProperties = new AMQP.BasicProperties().builder().correlationId(properties.getCorrelationId()).build();channel.basicPublish("", Constants.RPC_RESPONSE_QUEUE, basicProperties, response.getBytes());channel.basicAck(envelope.getDeliveryTag(), false);}};channel.basicConsume(Constants.RPC_REQUEST_QUEUE, false, consumer);}
}

客户端代码

import com.rabbitmq.client.*;
import rabbitmq.common.Constants;import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;/*** rpc 客户端* 1. 发送请求* 2. 接收响应*/
public class RpcClient {public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {//1. 建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT);connectionFactory.setUsername(Constants.USER_NAME);connectionFactory.setPassword(Constants.PASSWORD);connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();//2. 开启信道Channel channel = connection.createChannel();// 用来接收请求的队列channel.queueDeclare(Constants.RPC_REQUEST_QUEUE, true, false, false, null);// 用来发送相应的队列channel.queueDeclare(Constants.RPC_RESPONSE_QUEUE, true, false, false, null);//3. 发送请求String msg = "hello rpc...";//设置请求的唯一标识String correlationID = UUID.randomUUID().toString();//设置请求的相关属性AMQP.BasicProperties props = new AMQP.BasicProperties().builder().correlationId(correlationID).replyTo(Constants.RPC_RESPONSE_QUEUE).build();channel.basicPublish("", Constants.RPC_REQUEST_QUEUE, props, msg.getBytes());//4. 接收响应//使用阻塞队列, 来存储响应信息final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String respMsg = new String(body);System.out.println("接收到回调消息: "+ respMsg);if (correlationID.equals(properties.getCorrelationId())){//如果correlationID校验一致response.offer(respMsg);}}};channel.basicConsume(Constants.RPC_RESPONSE_QUEUE, true, consumer);String result = response.take();System.out.println("[RPC Client 响应结果]:"+ result);}
}

---

发布确认

Publisher Confirms 模式是 RabbitMQ 提供的一种机制，用于确保消息可靠发送到 RabbitMQ 服务器。以下是这种模式的主要特点和工作流程：

1. 开启 Confirm 模式:生产者通过调用 channel.confirmSelect() 将 Channel 设置为 Confirm 模式。在此模式下，每条发布的消息都会获得一个唯一的 ID，生产者可以将这些序列号与消息关联，以便跟踪消息状态。
2. 服务器确认:当消息被 RabbitMQ 服务器接收并处理后，服务器会异步地向生产者发送确认（ACK），确认中包含消息的唯一 ID，表明消息已经成功送达。
3. 确保消息可靠性:通过 Publisher Confirms 模式，生产者可以确保消息被 RabbitMQ 服务器成功接收，从而避免消息丢失的问题。

适用场景:
对数据安全性要求较高的场景: 例如金融交易、订单处理等需要确保数据不丢失的应用场景。

消息中间件在实际应用中可能会面临消息丢失的问题，这些问题可以大致分为三种情况：

生产者问题：由于应用程序故障、网络抖动等原因，生产者未能成功将消息发送到Broker。

消息中间件自身问题：生产者成功将消息发送到Broker，但Broker未能妥善保存消息，从而导致消息丢失。

消费者问题：Broker将消息发送给消费者后，消费者在处理消息时发生故障或未正确处理，导致Broker将消费失败的消息从队列中删除。

使用发送确认机制时，必须将信道设置为confirm（确认）模式。发布确认是 AMQP 0.9.1 协议的扩展，默认情况下不会启用。生产者可以通过 channel.confirmSelect() 将信道设置为确认模式。

其中确认模式有三种策略:

• 单独确认：每个消息都需要单独确认，生产者在发布每个消息后等待确认，确认消息会带有该消息的唯一ID。
• 批量确认：一次性确认多个消息，生产者通过设置multiple参数来确认某个序号之前的所有消息，这减少了确认的次数，提高了效率。
• 异步确认：生产者不必等待每个消息的确认，而是通过回调或其他机制异步接收确认，允许生产者继续处理其他任务，提高了整体性能。

当生产者将信道设置为confirm（确认）模式时：

生产者发布的所有消息将被分配一个唯一的ID（从1开始）。
一旦消息被投递到所有匹配的队列，RabbitMQ将向生产者发送一个确认，包含该消息的唯一ID。
只有在消息写入磁盘后，RabbitMQ才会发出确认（如果消息和队列是可持久化的）。
确认消息中的deliveryTag表示确认消息的序号。

代码实现

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import rabbitmq.common.Constants;import java.io.IOException;
import java.util.Collections;
import java.util.SortedSet;
import java.util.TreeSet;public class PublisherConfirms {private static final Integer MESSAGE_COUNT = 200;static Connection createConnection() throws Exception {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constants.HOST);connectionFactory.setPort(Constants.PORT);connectionFactory.setUsername(Constants.USER_NAME);connectionFactory.setPassword(Constants.PASSWORD);connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);return connectionFactory.newConnection();}public static void main(String[] args) throws Exception {//单独确认publishingMessagesIndividually();//批量确认publishingMessagesInBatches();//异步确认handlingPublisherConfirmsAsynchronously();}/*** 异步确认*/private static void handlingPublisherConfirmsAsynchronously() throws Exception{try (Connection connection = createConnection()){//1. 开启信道Channel channel = connection.createChannel();//2. 设置信道为confirm模式channel.confirmSelect();//3. 声明队列channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE3, true, false, false, null);//4. 监听confirm//集合中存储的是未确认的消息IDlong start = System.currentTimeMillis();SortedSet<Long> confirmSeqNo = Collections.synchronizedSortedSet(new TreeSet<>());channel.addConfirmListener(new ConfirmListener() {@Overridepublic void handleAck(long deliveryTag, boolean multiple) throws IOException {if (multiple){confirmSeqNo.headSet(deliveryTag+1).clear();}else {confirmSeqNo.remove(deliveryTag);}}@Overridepublic void handleNack(long deliveryTag, boolean multiple) throws IOException {if (multiple){confirmSeqNo.headSet(deliveryTag+1).clear();}else {confirmSeqNo.remove(deliveryTag);}//业务需要根据实际场景进行处理, 比如重发, 此处代码省略}});//5. 发送消息for (int i = 0; i < MESSAGE_COUNT; i++) {String msg = "hello publisher confirms"+i;long seqNo = channel.getNextPublishSeqNo();channel.basicPublish("",Constants.PUBLISHER_CONFIRMS_QUEUE3, null, msg.getBytes());confirmSeqNo.add(seqNo);}while (!confirmSeqNo.isEmpty()){Thread.sleep(10);}long end = System.currentTimeMillis();System.out.printf("异步确认策略, 消息条数: %d, 耗时: %d ms \n",MESSAGE_COUNT, end-start);}}/*** 批量确认*/private static void publishingMessagesInBatches() throws Exception{try(Connection connection = createConnection()) {//1. 开启信道Channel channel = connection.createChannel();//2. 设置信道为confirm模式channel.confirmSelect();//3. 声明队列channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE2, true, false, false, null);//4. 发送消息, 并进行确认long start = System.currentTimeMillis();int batchSize = 100;int outstandingMessageCount = 0;for (int i = 0; i < MESSAGE_COUNT; i++) {String msg = "hello publisher confirms"+i;channel.basicPublish("",Constants.PUBLISHER_CONFIRMS_QUEUE2, null, msg.getBytes());outstandingMessageCount++;if (outstandingMessageCount==batchSize){channel.waitForConfirmsOrDie(5000);outstandingMessageCount = 0;}}if (outstandingMessageCount>0){channel.waitForConfirmsOrDie(5000);}long end = System.currentTimeMillis();System.out.printf("批量确认策略, 消息条数: %d, 耗时: %d ms \n",MESSAGE_COUNT, end-start);}}/*** 单独确认*/private static void publishingMessagesIndividually() throws Exception {try(Connection connection = createConnection()) {//1. 开启信道Channel channel = connection.createChannel();//2. 设置信道为confirm模式channel.confirmSelect();//3. 声明队列channel.queueDeclare(Constants.PUBLISHER_CONFIRMS_QUEUE1, true, false, false, null);//4. 发送消息, 并等待确认long start = System.currentTimeMillis();for (int i = 0; i < MESSAGE_COUNT; i++) {String msg = "hello publisher confirms"+i;channel.basicPublish("",Constants.PUBLISHER_CONFIRMS_QUEUE1, null, msg.getBytes());//等待确认channel.waitForConfirmsOrDie(5000);}long end = System.currentTimeMillis();System.out.printf("单独确认策略, 消息条数: %d, 耗时: %d ms \n",MESSAGE_COUNT, end-start);}}
}

对比不同的消息数量