RabbitMQ官方案例学习记录

官方文档：RabbitMQ教程 — RabbitMQ (rabbitmq.com)

一、安装RabbitMQ服务

直接使用docker在服务器上安装

docker run -it -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3.12-management

安装完成后，访问15672端口，默认用户名和密码都是 guest，即可进入 

二、hello world——梦开始的地方

1. 介绍

        RabbitMQ 是一个消息代理：它接受并转发消息。 你可以把它想象成一个邮局：当你把你想要邮寄的邮件放在邮箱里时， 您可以确定，邮递员最终会将邮件递送给您的收件人。 在这个类比中，RabbitMQ是一个邮政信箱，一个邮局和一个信件载体。RabbitMQ和邮局的主要区别在于它不处理纸张， 相反，它接受、存储和转发数据的二进制 blob - 消息。

2. 一些术语

生产者：消息的发送方

队列queue：本质上是一个大型的消息缓冲区

消费者：消息的使用方

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

3. 编写代码

用 Java 编写两个程序;一个发送单个消息的生产者，一个接收的使用者并将消息打印出来。

（1）消息生产者：

编码过程：

先创建连接工厂，然后通过工厂创建连接，再通过连接创建channel。通过channel来绑定队列或者交换机，再用channel来生产或者消费消息。

channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());

对于这行代码，可以看到消息是根据QUEUE_NAME路由到对应的队列。

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
public class MQProducer {//设置队列名private final static String QUEUE_NAME = "hello-zy";public static void main(String[] argv) throws Exception {//创建连接工厂ConnectionFactory factory = new ConnectionFactory();//设置连接的服务器IP和端口号factory.setHost("123.249.112.12");factory.setPort(5672);//创建一个连接和通道，这里使用 try-with-resources语句，// 因为Connection和Channel都实现了java.lang.AutoCloseable，不需要在代码中再显式的关闭他们try(Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {/*创建队列 channel.queueDeclare()用于声明一个队列queue（队列名称）：指定要声明的队列的名称。durable（持久化）：指定队列是否是持久化的。当 RabbitMQ 重新启动时，持久化的队列将被保留下来。如果将该参数设置为 true，则队列将被持久化；如果设置为 false，则队列不会被持久化。注意，这里指的是队列本身的持久化，而不是队列中的消息。exclusive（排他性）：指定队列是否是排他的。如果将该参数设置为 true，则该队列只能被当前连接的消费者使用，并且在连接关闭时会自动删除该队列。如果设置为 false，则队列可供多个消费者使用。autoDelete（自动删除）：指定队列在不再被使用时是否自动删除。如果将该参数设置为 true，则当队列不再被消费者使用时，将自动删除该队列。如果设置为 false，则队列不会自动删除。arguments（参数）：指定队列的其他属性和参数，以键值对的形式提供。*/channel.queueDeclare(QUEUE_NAME, false, false, false, null);//发送消息String msg = "hello,world! RabbitMQ!";//channel.basicPublish("", QUEUE_NAME, null, msg.getBytes())// 这行代码的作用是将 msg 消息发布到默认交换器（空字符串）并使用QUEUE_NAME作为路由键，// 消息的属性设置为默认值，消息的内容为 msg.getBytes()，即将 msg 转换为字节数组后发送。channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());System.out.println(" 生产者发送消息：'" + msg + "'");}}
}

 然后运行这个生产者代码，去网页端查看：

（2）消息消费者：

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class MQConsumer {//声明队列 和消息发送方保持一致private final static String QUEUE_NAME = "hello-zy";public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory = new ConnectionFactory();//设置rabbitMQ服务端ipfactory.setHost("123.249.112.12");//这里不用try-with-resources 因为消费方需要一致保持监听，不要关闭Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);System.out.println(" 等待接收消息，退出请按 CTRL+C");//创建了一个消费者对象并实现了 handleDelivery() 方法作为回调方法。当消费者收到消息时，将自动执行该方法。DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" 消费了消息：'" + message + "'");};channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });}
}

去网页端查看消息是否被消费，可以看到多了一个消费者，并且消息已经没了。 

三、工作队列 WorkQueue

        工作队列（又名：任务队列）背后的主要思想是避免立即执行资源密集型任务，必须等待它要完成。相反，我们将任务安排在以后完成。我们将任务封装为消息并将其发送到队列。正在运行的工作进程 在后台将弹出任务并最终执行工作。当您运行许多工作线程时，任务将在它们之间共享。

WorkQueue的模型跟前面第一个案例Hello，World！的模型，最明显的区别其实就是，第一个案例他只有一个消费者。我们知道RabbitMQ他的消息是阅完即焚，即消费者一旦接收，这个消息直接就从Queue中被弹出了。
而现在这个案例，他有两个消费者（画两个只是方便，他当然也可以有3个、4个），他的消息应该是通过某种算法做负载均衡送到不同的消费者，让消费者进行处理，让消息不至于处理不过来，从而导致滞留在Queue中的消息被弹出。

思路如下：
1、我们先让Publish服务每秒发布50条消息到 simple.queue，来演示消息的频繁发送。
2、在Consumer服务中定义两个消费者，来监听我们的 simple.queue队列。
3、消费者1每秒处理40条消息，消费者2每秒处理30条消息。

1. 循环调度

（1）生产者：

生成50条消息：

public class Send {//设置队列名private final static String QUEUE_NAME = "hello-zy";public static void main(String[] argv) throws Exception {//创建连接ConnectionFactory factory = new ConnectionFactory();//设置连接的服务器IP和端口号factory.setHost("123.249.112.12");factory.setPort(5672);//创建一个通道，这里使用 try-with-resources语句，// 因为Connection和Channel都实现了java.lang.AutoCloseable，不需要在代码中再显式的关闭他们try(Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {channel.queueDeclare(QUEUE_NAME, false, false, false, null);//发送消息for(int i = 1; i <= 50; i++) {String msg = "hello, I am";msg = msg + i;channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());Thread.sleep(200);System.out.println(" 生产者发送消息：'" + msg + "'");}}}
}

（2）消费者1和消费者2

public class Worker_01 {//声明队列 和消息发送方保持一致private final static String QUEUE_NAME = "hello-zy";public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory = new ConnectionFactory();//设置rabbitMQ服务端ipfactory.setHost("123.249.112.12");//这里不用try-with-resources 因为消费方需要一致保持监听，不要关闭Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);System.out.println(" 等待接收消息，退出请按 CTRL+C");//创建了一个消费者对象并实现了 handleDelivery() 方法作为回调方法。当消费者收到消息时，将自动执行该方法。DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" Worker_01消费了消息：'" + message + "'");};channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });}
}public class Worker_02 {//声明队列 和消息发送方保持一致private final static String QUEUE_NAME = "hello-zy";public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory = new ConnectionFactory();//设置rabbitMQ服务端ipfactory.setHost("123.249.112.12");//这里不用try-with-resources 因为消费方需要一致保持监听，不要关闭Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);System.out.println(" 等待接收消息，退出请按 CTRL+C");//创建了一个消费者对象并实现了 handleDelivery() 方法作为回调方法。当消费者收到消息时，将自动执行该方法。DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" Worker_02消费了消息：'" + message + "'");};channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });}
}

启动两个消费者，然后开始生产消息，控制台打印如下：

 可以看到是交替消费的：

2. 消息确认

        在上面的代码中，一旦消息传递给消费者，就会立即被删除。如果在消费过程中，消费者宕机，消息没消费成功，但是因为已经投递出去了，消息从队列删掉。就会出现：消息未消费，且丢失的问题。对于这种情况，我们希望没消费成功的消息，转交给其他的消费者消费。

        为了确保消息永远不会丢失，RabbitMQ 支持消息确认。确认由消费者告诉 RabbitMQ 已收到特定消息，并处理，RabbitMQ可以自由删除它。

        如果消费者没有发送确认消息，rabbitMQ可以知道消息没有消费成功，并将重新排队。

        在消费者返回确认消息时强制实施超时（默认为 30 分钟）。 这有助于检测从不确认交付的错误（卡住）消费者。 可以按照传递确认超时中所述增加此超时。

        默认情况下，手动消息确认处于打开状态。在上一个 示例，我们通过 autoAck=true 标志明确关闭了它们。是时候将此标志设置为 false ，来使消费者发送确认消息。

（1）消费者

生产者还是不变，生成10条消息到队列中：

创建消费者，启动手动确认：

public class Consumer {private final static String QUEUE_NAME = "hello-zy";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("123.249.112.12");factory.setPort(5672);Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 将autoAck参数设置为false，关闭自动消息确认/** new DefaultConsumer(channel) { ... }：这是一个匿名内部类，用于定义消息处理的逻辑。* 它继承自DefaultConsumer，并覆盖了 handleDelivery 方法，以自定义消息的处理方式。* */channel.basicConsume(QUEUE_NAME, false, new DefaultConsumer(channel) {/*handleDelivery 方法：这是 DefaultConsumer 类中的方法，用于处理从队列中接收的消息。consumerTag：标识消费者的标签。envelope：包含与消息相关的元数据，如交付标签、交付模式等。properties：包含消息的属性，如消息的头部信息。body：消息的内容，以字节数组形式提供。*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String message = new String(body, "UTF-8");System.out.println("消费者接收消息: '" + message + "'");// 在消息处理成功后，发送确认消息channel.basicAck(envelope.getDeliveryTag(), false);System.out.println("确认收到了消息："+message);}});}
}

启动消费者：

（2）验证未成功消费情况

先生产6条消息到队列：

简单修改一下消费者代码，在中间添加判断逻辑，当碰到消息5的时候，退出消费者：

public class Consumer1 {private final static String QUEUE_NAME = "hello-zy";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("123.249.112.12");Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);try {// 将autoAck参数设置为false，关闭自动消息确认channel.basicConsume(QUEUE_NAME, false, new DefaultConsumer(channel) {@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)throws IOException {String message = new String(body, "UTF-8");System.out.println("消费者接收消息: '" + message + "'");// 模拟某个条件，例如消息处理成功if (!message.contains("5")) {// 在消息处理成功后，发送确认消息channel.basicAck(envelope.getDeliveryTag(), false);System.out.println("确认收到了消息：" + message);}// 模拟消费者退出if (message.contains("5")) {throw new RuntimeException("Consumer exiting...");}}});} catch (RuntimeException e) {// 当消费者退出时，捕获异常System.out.println("Consumer exited.");}}
}

 

可以看到，当消费者消费到第五条消息后，因为抛出了异常，所以后面的消息都未消费成功，所以会把第五条消息和第六条消息再放回消息队列，查看消息队列，可以看到还有两条消息在队列中：

如果把参数改成true，也就是自动确认：这就会导致未消费的消息丢失

 

3. 消息持久性

我们已经保证了消息未消费的情况下不会丢失，但是如果RabbitMQ服务器宕机，消息还是会丢失。这就涉及到一个消息持久化的问题。

需要做两件事来确保 消息不会丢失：我们需要将队列和消息都标记为durable。 （重新定义一个队列）

此时，我们确信durable_queue队列不会丢失 即使 RabbitMQ 重新启动。现在我们需要将我们的消息标记为持久 - 通过设置消息属性（实现基本属性） 到值PERSISTENT_TEXT_PLAIN。 

 

如果不设置未持久性，重启docker的rabbitmq容器，队列消息就丢了，设置durable=true后， 即使重启docker容器，队列和消息都不会丢失。

四、发布/订阅模式

        在前面，都是一条消息由一个消费者消费。如果一条消息需要被多个消费者消费，那么就需要引入发布/订阅模式。

1. 交换机

关于交换机的概念（）：

一个生产者给 多个 队列发消息，1 个生产者对多个队列。
交换机的作用：提供消息转发功能，类似于网络路由器
要解决的问题：怎么把消息转发到不同的队列上，好让消费者从不同的队列消费。

交换机有多种类别：fanout、direct, topic, headers

fanout（扇出）

扇出、广播
特点：消息会被转发到所有绑定到该交换机的队列
场景：很适用于发布订阅的场景。比如写日志，可以多个系统间共享

Direct 直接

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

特点：消息会根据路由键转发到指定的队列
场景：特定的消息只交给特定的系统（程序）来处理
绑定关系：完全匹配字符串

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

 

 topic 交换机

特点：消息会根据一个 模糊的 路由键转发到指定的队列
场景：特定的一类消息可以交给特定的一类系统（程序）来处理
绑定关系：可以模糊匹配多个绑定
●*：匹配一个单词，比如 *.orange，那么 a.orange、b.orange 都能匹配
●#：匹配 0 个或多个单词，比如 a.#，那么 a.a、a.b、a.a.a 都能匹配

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

Headers 交换机 

类似主题和直接交换机，可以根据 headers 中的内容来指定发送到哪个队列。使用消息头headers来路由消息。

 2. 使用fanout交换机来实现发布/订阅

（1）生产者

public class LogProducer {private final static String EXCHANGE_NAME = "logs";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("123.249.112.12");factory.setPort(5672);try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {//声明交换机类型为：FANOUTchannel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);for (int i = 1; i <= 5; i++) {String message = "Log message " + i;//消息发到交换机中，而不是像之前点对点那样直接发到消息队列中channel.basicPublish(EXCHANGE_NAME, "", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());System.out.println("生产者发送日志: " + message);}}}
}

查看交换机列表，可以看到新增了一个名为logs的交换机，类型为 fanout：

 （2）消费者

创建两个消费者，都绑定一个交换机，名字为log

public class LogConsumer {private final static String EXCHANGE_NAME = "logs";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("123.249.112.12");Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);// 创建一个临时队列String queueName = channel.queueDeclare().getQueue();// 将队列绑定到交换器channel.queueBind(queueName, EXCHANGE_NAME, "");System.out.println("等待接收日志消息...");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println("消费者1接收消息: '" + message + "'");};channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });}
}
public class LogConsumer2 {private final static String EXCHANGE_NAME = "logs";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("123.249.112.12");Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);// 创建一个临时队列String queueName = channel.queueDeclare().getQueue();// 将队列绑定到交换器channel.queueBind(queueName, EXCHANGE_NAME, "");System.out.println("等待接收日志消息...");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println("消费者2接收消息: '" + message + "'");};channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});}
}

 启动两个消费者，然后启动生产者发送消息，可以看到两个消费者都消费了消息：

五、路由

在上面的例子中，我们实现了一个简单的日志记录系统。能够向多个消费者广播消息。

下面要增加一个新的功能：

        例如，我们将只能将关键错误消息定向到 日志文件（以节省磁盘空间），同时仍然能够打印所有控制台上的日志消息。

1. 绑定队列和交换机

之前是这样绑定的：第三个参数就是路由键

channel.queueBind(queueName, EXCHANGE_NAME, "");

绑定是交换和队列之间的关系。

        绑定可以采用额外的路由键参数。为了避免 与basic_publish参数混淆，我们将它称为绑定键。绑定键的含义取决于交换类型。

channel.queueBind(queueName, EXCHANGE_NAME, "black");

2. 使用direct交换机绑定

（1）生产者

模拟三条不同的消息，指定消息1的路由键为orange，消息2的路由键为black，消息3的路由键为green。

public class LogProducer {private final static String EXCHANGE_NAME = "direct_logs";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("123.249.112.12");factory.setPort(5672);try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {//声明交换机类型为 directchannel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);String message1 = "Message with routing key orange";String message2 = "Message with routing key black";String message3 = "Message with routing key green";// 发布消息到交换器，并指定不同的路由键channel.basicPublish(EXCHANGE_NAME, "orange", null, message1.getBytes());channel.basicPublish(EXCHANGE_NAME, "black", null, message2.getBytes());channel.basicPublish(EXCHANGE_NAME, "green", null, message3.getBytes());System.out.println("生产者发送消息完成.");}}
}

（2）消费者

创建两个消费者

消费者1和消费者2都绑定对应的交换机，其中消费者1对应路由键orange，消费者2对应路由键black和orange。

public class LogConsumer1 {private final static String EXCHANGE_NAME = "direct_logs";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("123.249.112.12");factory.setPort(5672);Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);// 创建一个临时队列String queueName = channel.queueDeclare().getQueue();// 绑定队列到交换器，指定路由键为 "orange"channel.queueBind(queueName, EXCHANGE_NAME, "orange");System.out.println("等待接收 orange 消息...");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println("消费者1接收消息: '" + message + "'");};channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});}
}
public class LogConsumer2 {private final static String EXCHANGE_NAME = "direct_logs";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("123.249.112.12");factory.setPort(5672);Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);// 创建一个临时队列String queueName = channel.queueDeclare().getQueue();// 绑定队列到交换器，指定路由键为 "black" 和 "green"channel.queueBind(queueName, EXCHANGE_NAME, "black");channel.queueBind(queueName, EXCHANGE_NAME, "green");System.out.println("等待接收 black 和 green 消息...");DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println("消费者2接收消息: '" + message + "'");};channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {});}
}