消息中间件之RabbitMQ的安装及消息发送接收

喜欢ヅ旅行 2022-10-12 12:11 183阅读 0赞

1、MQ简述

1.1、什么是MQ

MQ(message queue)，从字面意思上看，本质是个队列，FIFO 先入先出，只不过队列中存放的内容是 message 而已，还是一种跨进程的通信机制，用于上下游传递消息。在互联网架构中，MQ 是一种非常常 见的上下游“逻辑解耦+物理解耦”的消息通信服务。使用了 MQ 之后，消息发送上游只需要依赖 MQ，不 用依赖其他服务。

1.2、为什么要用MQ

*  **流量消峰**  
    举个例子，如果订单系统最多能处理一万次订单，这个处理能力应付正常时段的下单时绰绰有余，正常时段我们下单一秒后就能返回结果。但是在高峰期，如果有两万次下单操作系统是处理不了的，只能限制订单超过一万后不允许用户下单。使用消息队列做缓冲，我们可以取消这个限制，把一秒内下的订单分散成一段时间来处理，这时有些用户可能在下单十几秒后才能收到下单成功的操作，但是比不能下单的体验要好。
 *  **应用解耦**  
    以电商应用为例，应用中有订单系统、库存系统、物流系统、支付系统。用户创建订单后，如果耦合 调用库存系统、物流系统、支付系统，任何一个子系统出了故障，都会造成下单操作异常。当转变成基于 消息队列的方式后，系统间调用的问题会减少很多，比如物流系统因为发生故障，需要几分钟来修复。在 这几分钟的时间里，物流系统要处理的内存被缓存在消息队列中，用户的下单操作可以正常完成。当物流 系统恢复后，继续处理订单信息即可，中单用户感受不到物流系统的故障，提升系统的可用性。
 *  **异步处理**  
    有些服务间调用是异步的，例如 A 调用 B，B 需要花费很长时间执行，但是 A 需要知道 B 什么时候可以执行完。以前一般有两种方式，A 过一段时间去调用 B 的查询 api 查询。或者 A 提供一个 callback api， B 执行完之后调用 api 通知 A 服务。这两种方式都不是很优雅。使用消息总线，可以很方便解决这个问题， A 调用 B 服务后，只需要监听 B 处理完成的消息，当 B 处理完成后，会发送一条消息给 MQ，MQ 会将此消息转发给 A 服务。这样 A 服务既不用循环调用 B 的查询 api，也不用提供 callback api。同样B 服务也不用 做这些操作。A 服务还能及时的得到异步处理成功的消息。

1.3、MQ 的选择

*  Kafka  
    Kafka 主要特点是基于Pull 的模式来处理消息消费，追求高吞吐量，一开始的目的就是用于日志收集 和传输，适合产生大量数据的互联网服务的数据收集业务。大型公司建议可以选用，如果有日志采集功能， 肯定是首选 kafka 了。
 *  RocketMQ  
    天生为金融互联网领域而生，对于可靠性要求很高的场景，尤其是电商里面的订单扣款，以及业务削 峰，在大量交易涌入时，后端可能无法及时处理的情况。RoketMQ 在稳定性上可能更值得信赖，这些业务 场景在阿里双 11 已经经历了多次考验，如果你的业务有上述并发场景，建议可以选择 RocketMQ。
 *  RabbitMQ  
    结合 erlang 语言本身的并发优势，性能好时效性微秒级，社区活跃度也比较高，管理界面用起来十分 方便，如果你的数据量没有那么大，中小型公司优先选择功能比较完备的 RabbitMQ。

2、RabbitMQ简述

2.1、RabbitMQ是什么

**RabbitMQ 是一个消息中间件：它接受并转发消息。**

你可以把它当做一个快递站点，当你要发送一个包裹时，你把你的包裹放到快递站，快递员最终会把你的快递送到收件人那里，按照这种逻辑 RabbitMQ 是 一个快递站，一个快递员帮你传递快件。

RabbitMQ 与快递站的主要区别在于，它不处理快件而是接收， 存储和转发消息数据。

2.2、四大核心概念

*  生产者  
    产生数据发送消息的程序是生产者
 *  交换机  
    交换机是 RabbitMQ 非常重要的一个部件，一方面它接收来自生产者的消息，另一方面它将消息 推送到队列中。交换机必须确切知道如何处理它接收到的消息，是将这些消息推送到特定队列还是推 送到多个队列，亦或者是把消息丢弃，这个得有交换机类型决定
 *  队列  
    队列是 RabbitMQ 内部使用的一种数据结构，尽管消息流经 RabbitMQ 和应用程序，但它们只能存 储在队列中。队列仅受主机的内存和磁盘限制的约束，本质上是一个大的消息缓冲区。许多生产者可 以将消息发送到一个队列，许多消费者可以尝试从一个队列接收数据。这就是我们使用队列的方式
 *  消费者  
    消费与接收具有相似的含义。消费者大多时候是一个等待接收消息的程序。请注意生产者，消费 者和消息中间件很多时候并不在同一机器上。同一个应用程序既可以是生产者又是可以是消费者。

3、RabbitMQ的安装

这里就直接使用虚拟机进行安装，首先是安装erlang对应的 yum repo（在这里使用的是**centos8** 版本的虚拟机）

curl -s https://packagecloud.io/install/repositories/rabbitmq/erlang/script.rpm.sh | sudo bash

之后执行命令进行安装erlang

yum install erlang

安装erlang之后，可以使用`erl`进行验证  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70]  
随后安装rabbitmq

curl -s https://packagecloud.io/install/repositories/rabbitmq/rabbitmq-server/script.rpm.sh | sudo bash

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70 1]  
进行安装rabbitmq

yum -y install rabbitmq-server

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70 2]  
之后我们对rabbitmq进行一些设置。  
首先设置RABBITMQ开机启动：`chkconfig rabbitmq-server on`  
之后我们启动服务：`systemctl start rabbitmq-server.service`  
启动之后我们进行查看rabbitmq的状态：`service rabbitmq-server status`  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70 3]  
还需要安装一个web管理界面，而在安装的过程中需要将服务先停掉，使用命令`systemctl stop rabbitmq-server.service`，再查看一下状态，状态就变成了一个loaded加载中。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70 4]  
安装插件，使用命令：`rabbitmq-plugins enable rabbitmq_management`  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70 5]  
插件安装好了之后我们再次的启动服务，之后我们在浏览器当中进行访问，IP地址+15672  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70 6]  
在这里，可能会出现访问不了，访问不了的原因基本上是虚拟机的防火墙没有关，我们对虚拟机的防火墙进行关掉。  
查看状态：`systemctl status firewalld`  
关闭防火墙：`systemctl status firewalld`

之后我们使用guest进行登录是登录不上去的，我们需要进行创建用户进行登录，我们可以先看一下有哪些账户。  
使用命令：`rabbitmqctl list_users`  
![在这里插入图片描述][35a54fba54dc4e22bffde0170f123e37.png]  
之后使用命令创建用户：`rabbitmqctl add_user root 123456`  
给用户设置角色：`rabbitmqctl set_user_tags admin administrator`  
给用户设置权限：`rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"`

到这里就设置完了，之后我们使用root 123456进行登录即可。

4、第一个HelloWord程序

我们创建一个maven项目，创建完成之后在pom.xml文件当中添加一下所需依赖。

<build>
            <plugins>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <configuration>
                        <source>8</source>
                        <target>8</target>
                    </configuration>
                </plugin>
            </plugins>
        </build>
        <dependencies>
            
            <dependency>
                <groupId>com.rabbitmq</groupId>
                <artifactId>amqp-client</artifactId>
                <version>5.8.0</version>
            </dependency>
            
            <dependency>
                <groupId>commons-io</groupId>
                <artifactId>commons-io</artifactId>
                <version>2.6</version>
            </dependency>
        </dependencies>

之后我们新建一个生产者的类，用来发送消息。

package com.lzq.rabbitmq.one;
    
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    import java.net.InetAddress;
    import java.util.Map;
    
    public class Producer { 
        // 消息队列
        private final static String QUEUE_NAME="hello";
    
        // 发送消息
        public static void main(String[] args) throws Exception{ 
            // 创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            // 工厂IP
            factory.setHost("192.168.101.128");
            // 用户名 密码
            factory.setUsername("root");
            factory.setPassword("123456");
    
            // 创建连接
            Connection connection = factory.newConnection();
    
            // 获取信道
            Channel channel =  connection.createChannel();
    
            /** * 生成队列 * 参数说明 * 1、队列名称 * 2、队列消息是否持久化（存储在磁盘上）默认不持久化 * 3、队列是否只能由一个消费者使用，true表示运行多个消费者共享， * 4、是否自动删除，最后一个消费者断开连接是否自动删除 * 5、其他参数 * */
            channel.queueDeclare(QUEUE_NAME,true,false,false,null);
    
            // 消息内容
            String message = "hello world";
    
            /** * 发送消息，通过信道发送消息 * 参数说明 * 1、交换机 * 2、消息队列 * 3、其他参数 * 4、消息体 * */
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            System.out.println("发送成功");
        }
    }

运行Java代码发送消息之后，可以在web管理当中查看消息队列。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70 7]  
之后新建一个消费者对消息进行接收。在前面的生产者进行发送消息给到消息队列之后，在消费者当中获取到这个消息队列即对该消息进行接收。

package com.lzq.rabbitmq.one;
    
    import com.rabbitmq.client.*;
    
    public class Consumer { 
        // 消息队列
        private final static String QUEUE_NAME = "hello";
    
        public static void main(String[] args) throws Exception { 
            // 创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.101.128");
            factory.setUsername("root");
            factory.setPassword("123456");
            Connection connection = factory.newConnection();
    
            // 获取信道
            Channel channel = connection.createChannel();
    
            /** * 消费消息 * 参数说明 * 1、队列 * 2、消费之后 自动应答 true 手动应答 false * 3、消费者未成功消费的回调 * 4、消费者取消消费的回调 * */
            // 回调函数的声明
            DeliverCallback deliverCallback = (consumerTag, message) -> { 
                System.out.println(new String(message.getBody()));
            };
            CancelCallback cancelCallback = (consumerTag) -> { 
                System.out.println("消费中断");
            };
            channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
        }
    }

5、work Queues 工作队列

在这里前面的第一个 helloworld 的代码其中生产者和消费者都需要进行连接到连接工厂并且在工厂当中获取信道，在这里我们这一段代码就可以进行抽离出来，作为一个共用的方法，也是避免了代码的冗余。

public class RabbitmqUtil { 
        public static Channel getChannel() throws Exception{ 
            // 创建连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("192.168.101.128");
            factory.setUsername("root");
            factory.setPassword("123456");
            Connection connection = factory.newConnection();
    
            // 获取信道
            Channel channel = connection.createChannel();
            return channel;
        }
    }

工作队列(又称任务队列)的主要思想是避免立即执行资源密集型任务，而不得不等待它完成。相反我们安排任务在之后执行。我们把任务封装为消息并将其发送到队列。在后台运行的工作进程将弹出任务并最终执行作业。当有多个工作线程时，这些工作线程将一起处理这些任务。

5.1、轮训分发消息

之后我们相同的创建生产者和消费者进行测试轮询，也就是使用两个消费者（代码一致，分别打印出一个c1和c2作为消费者的名称，将两个消费者run起来）

之后同样的创建一个生产者用来发送消息，查看两个消费者的消息如何进行接受消息进行处理。

public static void main(String[] args) throws Exception { 
            Channel channel = RabbitmqUtil.getChannel();
    
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
    
            Scanner sc = new Scanner(System.in);
            while (sc.hasNext()) { 
                // 消息内容
                String message = sc.next();
                channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
                System.out.println("发送成功");
            }
        }

在生产者当中进行发送消息之后，在两个消费者当中会一个接收一条消息，从此往复，这也就是轮训分发消息。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70 8]  
5.2、消息应答

消费者完成一个任务可能需要一段时间，如果其中一个消费者处理一个长的任务并仅只完成了部分突然它挂掉了，会发生什么情况。RabbitMQ 一旦向消费者传递了一条消息，便立即将该消息标记为删除。在这种情况下，突然有个消费者挂掉了，我们将丢失正在处理的消息。以及后续发送给该消费这的消息，因为它无法接收到。

为了保证消息在发送过程中不丢失，rabbitmq 引入消息应答机制，**消息应答就是：消费者在接收到消息并且处理该消息之后，告诉 rabbitmq 它已经处理了，rabbitmq 可以把该消息删除了。**

**自动应答机制**

消息发送后立即被认为已经传送成功，这种模式需要在高吞吐量和数据传输安全性方面做权衡,因为这种模式如果消息在接收到之前，消费者那边出现连接或者 channel 关闭，那么消息就丢失了，当然另一方面这种模式消费者那边可以传递过载的消息，没有对传递的消息数量进行限制，当然这样有可能使得消费者这边由于接收太多还来不及处理的消息，导致这些消息的积压，最终使得内存耗尽，最终这些消费者线程被操作系统杀死，**所以这种模式仅适用在消费者可以高效并以某种速率能够处理这些消息的情况下使用。**

**消息应答的方法**

*  .Channel.basicAck(用于肯定确认)  
    RabbitMQ 已知道该消息并且成功的处理消息，可以将其丢弃了
 *  Channel.basicNack(用于否定确认)
 *  Channel.basicReject(用于否定确认)  
    与 Channel.basicNack 相比少一个参数  
    不处理该消息了直接拒绝，可以将其丢弃了

**手动应答**

手动应答的好处是可以批量应答并且减少网络拥堵

boolean ackAuto = false;
            channel.basicConsume(TASK, ackAuto, deliverCallback, cancelCallback);

multiple 的 true 和 false 代表不同意思  
true 代表批量应答 channel 上未应答的消息  
比如说 channel 上有传送 tag 的消息 5,6,7,8 当前 tag 是 8 那么此时5-8 的这些还未应答的消息都会被确认收到消息应答  
false 同上面相比  
只会应答 tag=8 的消息 5,6,7 这三个消息依然不会被确认收到消息应答

**消息自动重新入队**

如果消费者由于某些原因失去连接(其通道已关闭，连接已关闭或 TCP 连接丢失)，导致消息未发送 ACK 确认，RabbitMQ 将了解到消息未完全处理，并将对其重新排队。如果此时其他消费者可以处理，它将很快将其重新分发给另一个消费者。这样，即使某个消费者偶尔死亡，也可以确保不会丢失任何消息。

**消息手动应答案例**

默认消息采用的是自动应答，所以我们要想实现消息消费过程中不丢失，需要把自动应答改为手动应答，在消费者当中使用线程对停留时间进行控制，对两个消费者分别停留1秒和30秒用来区分快速接收消息应答和需要较长时间才会做出应答，将自动应答改成手动应答。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70 9]  
然后这里的SleepUtil这个工具类当中也就是使用sleep进行停留。

public class SleepUtil { 
        public static void sleep(int second) { 
            try { 
                Thread.sleep(1000 * second);
            } catch (InterruptedException _ignored) { 
                Thread.currentThread().interrupt();
            }
        }
    }

我们使用两个消费者进行接收消息，由于消息是进行轮询分发的，在这里第二个消费者接收消息需要较长的时间进行等待，而当这里需要等待时间较长的服务宕机了，消息是不会丢失的，而这条消息会被另外一个消费者进行消费。

5.3、RabbitMQ持久化

在前面我们已经明白了如何处理任务不丢失的情况，但是如何保障当 RabbitMQ 服务停掉以后消息生产者发送过来的消息不丢失。默认情况下 RabbitMQ 退出或由于某种原因崩溃时，它忽视队列和消息，除非告知它不要这样做。确保消息不会丢失需要做两件事：**我们需要将队列和消息都标记为持久化。**

**队列如何实现持久化**

在创建的队列如果是非持久化的，rabbitmq 如果重启的话，该队列就会被删除掉，如果要队列实现持久化 需要在声明队列的时候把 durable 参数设置为持久化。

boolean durable = true;
    channel.queueDeclare(TASK, durable, false, false, null);

但是需要注意的就是如果之前声明的队列不是持久化的，需要把原先队列先删除，或者重新创建一个持久化的队列，不然就会报错。在控制台中持久化的UI显示区会加上一个D进行表示。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70 10]  
**消息实现持久化**

要想让消息实现持久化需要在消息生产者修改代码`MessageProperties.PERSISTENT_TEXT_PLAIN` 添加这个属性。

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

将消息标记为持久化并不能完全保证不会丢失消息。尽管它告诉 RabbitMQ 将消息保存到磁盘，但是这里依然存在当消息刚准备存储在磁盘的时候 但是还没有存储完，消息还在缓存的一个间隔点。此时并没有真正写入磁盘。持久性保证并不强，但是对于我们的简单任务队列而言，这已经绰绰有余了。

**不公平分发**

在最开始的时候我们学习到 RabbitMQ 分发消息采用的轮训分发，但是在某种场景下这种策略并不是很好，比方说有两个消费者在处理任务，其中有个消费者1处理任务的速度非常快，而另外一个消费者2处理速度却很慢，这个时候我们还是采用轮训分发的化就会到这处理速度快的这个消费者很大一部分时间处于空闲状态，而处理慢的那个消费者一直在干活，这种分配方式在这种情况下其实就不太好，但是RabbitMQ 并不知道这种情况它依然很公平的进行分发。为了避免这种情况，我们可以设置参数 `channel.basicQos(1);`

// 采用手动应答
            boolean ackAuto = false;
            // 不公平分发
            channel.basicQos(1);
            channel.basicConsume(TASK, ackAuto, deliverCallback, cancelCallback);

在前面的代码当中两个消费者分别等待的时间为1秒和30秒，那我们给两个消费者都加上不公平分发，启动程序查看一下效果。

意思就是如果这个任务我还没有处理完或者我还没有应答你，你先别分配给我，我目前只能处理一个任务，然后 rabbitmq 就会把该任务分配给没有那么忙的那个空闲消费者，当然如果所有的消费者都没有完成手上任务，队列还在不停的添加新任务，队列有可能就会遇到队列被撑满的情况，这个时候就只能添加新的 worker 或者改变其他存储任务的策略。

**预取值**

本身消息的发送就是异步发送的，所以在任何时候，channel 上肯定不止只有一个消息另外来自消费者的手动确认本质上也是异步的。因此这里就存在一个未确认的消息缓冲区，因此希望开发人员能限制此缓冲区的大小，以避免缓冲区里面无限制的未确认消息问题。这个时候就可以通过使用 basic.qos 方法设置“预取计数”值来完成的。该值定义通道上允许的未确认消息的最大数量。一旦数量达到配置的数量，RabbitMQ 将停止在通道上传递更多消息，除非至少有一个未处理的消息被确认，例如，假设在通道上有未确认的消息 5、6、7，8，并且通道的预取计数设置为 4，此时 RabbitMQ 将不会在该通道上再传递任何消息，除非至少有一个未应答的消息被 ack。比方说 tag=6 这个消息刚刚被确认 ACK，RabbitMQ 将会感知这个情况到并再发送一条消息。消息应答和 QoS 预取值对用户吞吐量有重大影响。通常，增加预取将提高向消费者传递消息的速度。虽然自动应答传输消息速率是最佳的，但是，在这种情况下已传递但尚未处理的消息的数量也会增加，从而增加了消费者的 RAM 消耗(随机存取存储器)应该小心使用具有无限预处理的自动确认模式或手动确认模式，消费者消费了大量的消息如果没有确认的话，会导致消费者连接节点的内存消耗变大，所以找到合适的预取值是一个反复试验的过程，不同的负载该值取值也不同 100 到 300 范围内的值通常可提供最佳的吞吐量，并且不会给消费者带来太大的风险。预取值为 1 是最保守的。当然这将使吞吐量变得很低，特别是消费者连接延迟很严重的情况下，特别是在消费者连接等待时间较长的环境中。对于大多数应用来说，稍微高一点的值将是最佳的。

6、发布确认

6.1、发布确认原理

生产者将信道设置成 confirm 模式，一旦信道进入 confirm 模式，所有在该信道上面发布的消息都将会被指派一个唯一的 ID(从 1 开始)，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者(包含消息的唯一 ID)，这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker 回传给生产者的确认消息中 delivery-tag 域包含了确认消息的序列号，此外 broker 也可以设置basic.ack 的 multiple 域，表示到这个序列号之前的所有消息都已经得到了处理。

confirm 模式最大的好处在于他是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果 RabbitMQ 因为自身内部错误导致消息丢失，就会发送一条 nack 消息，生产者应用程序同样可以在回调方法中处理该 nack 消息。

6.2、开启发布确认的方法

发布确认默认是没有开启的，如果要开启需要调用方法 confirmSelect，每当你要想使用发布确认，都需要在 channel 上调用该方法。

// 发布确认开启
            channel.confirmSelect();

6.3、单个发布确认

这是一种简单的确认方式，它是一种同步确认发布的方式，也就是发布一个消息之后只有它被确认发布，后续的消息才能继续发布,waitForConfirmsOrDie(long)这个方法只有在消息被确认的时候才返回，如果在指定时间范围内这个消息没有被确认那么它将抛出异常。

这种确认方式有一个最大的缺点就是:发布速度特别的慢，因为如果没有确认发布的消息就会阻塞所有后续消息的发布，这种方式最多提供每秒不超过数百条发布消息的吞吐量。当然对于某些应用程序来说这可能已经足够了。下面使用代码进行演示单个发布确认。

// 单个确认
        public static void batchConfirmation() throws Exception { 
            Channel channel = RabbitmqUtil.getChannel();
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName, true, false, false, null);
            // 发布确认开启
            channel.confirmSelect();
            // 开始时间
            long start = System.currentTimeMillis();
            // 批量发消息
            for (int i = 0; i < MESSAGE_COUNT; i++) { 
                String message = i + "";
                channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));
                // 消息确认
                boolean flag = channel.waitForConfirms();
                if (flag) { 
                    System.out.println("success");
                }
            }
            long end = System.currentTimeMillis();
            System.out.println("单个确认用时：" + (end - start));
        }

6.4、批量确认发布

上面那种方式非常慢，与单个等待确认消息相比，先发布一批消息然后一起确认可以极大地提高吞吐量，当然这种方式的缺点就是:当发生故障导致发布出现问题时，不知道是哪个消息出现问题了，我们必须将整个批处理保存在内存中，以记录重要的信息而后重新发布消息。当然这种方案仍然是同步的，也一样阻塞消息的发布。在这里我们只需要对上述代码的每一次发送的消息都会进行确认，而批量也就是只确认几次或者一次，如下代码所示：在消息都发送完成之后进行确认。所用时间比单个发布的更短，但是无法保证每一条消息发送成功与否。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70 11]  
6.4、异步确认发布

异步确认虽然编程逻辑比上两个要复杂，但是性价比最高，无论是可靠性还是效率都没得说，他是利用回调函数来达到消息可靠性传递的，这个中间件也是通过函数回调来保证是否投递成功。

public static void asyncConfirmation() throws Exception { 
            Channel channel = RabbitmqUtil.getChannel();
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName, true, false, false, null);
            // 发布确认开启
            channel.confirmSelect();
            // 开始时间
            long start = System.currentTimeMillis();
            // 消息监听器 哪些消息成功，哪些失败
            /** * 1、deliveryTag 消息标识 * 2、multiple 是否批量 * */
            ConfirmCallback ack = (deliveryTag, multiple) -> { 
    
            };
            ConfirmCallback nack = (deliveryTag, multiple) -> { 
                System.out.println("未确认 ：" + deliveryTag);
            };
            channel.addConfirmListener(ack, nack);
            // 批量发消息
            for (int i = 0; i < MESSAGE_COUNT; i++) { 
                String message = i + "";
                channel.basicPublish("", queueName, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));
            }
            long end = System.currentTimeMillis();
            System.out.println("异步确认用时：" + (end - start));
        }

6.5、如何处理异步未确认消息

最好的解决的解决方案就是把未确认的消息放到一个基于内存的能被发布线程访问的队列，比如说用 ConcurrentLinkedQueue（并发链式队列） 这个队列在 confirm callbacks 与发布线程之间进行消息的传递。

ConcurrentSkipListMap<Long,String> concurrentSkipListMap = new ConcurrentSkipListMap<>();
    	ConfirmCallback ack = (deliveryTag, multiple) -> { 
    		if (multiple){ 
    			ConcurrentNavigableMap<Long,String> map =  concurrentSkipListMap.headMap(deliveryTag);
    	        map.clear();
    		}else{ 
    			concurrentSkipListMap.remove(deliveryTag);
            }
        };

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70_pic_center]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70]: /images/20221005/af9d1fdb6f7049c59762f494f40d0f65.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70 1]: /images/20221005/09b6210802bb4dae862f7c3f5a92f052.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70 2]: /images/20221005/c7f91c4a563c469580edfcd657393507.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70 3]: /images/20221005/c9ab41056d214556bb75f0e613a74072.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70 4]: /images/20221005/2c8371ec0d564d05b28643a8375cf32a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70 5]: /images/20221005/d4df890de4104d5c9e6bde2c5ac6f668.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70 6]: /images/20221005/058318d22dde474d8c5b160c95ddb932.png
[35a54fba54dc4e22bffde0170f123e37.png]: /images/20221005/b1c6f9609063449286279d9ae00b3192.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70 7]: /images/20221005/a7d5f441b56b412fa8b54701ccbbeeb8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70 8]: /images/20221005/37a5d634e35144a698861290259ef3d4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70 9]: /images/20221005/9829595edaa14921964028cf4fa05942.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70 10]: /images/20221005/d304e13ee90e4c0e9ff6a6cfee75691d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70 11]: /images/20221005/5a4273910d0f4f13990bff09a57c4ddf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ0OTczMTU5_size_16_color_FFFFFF_t_70_pic_center]: /images/20221005/6ff100d7c42c434cb7551378f440bd2c.png