RabbitMQ 消息应答与发布
文章目录
前言
上文我们已经成功安装完成 rabbitmq 并且写一个helloworld 入门案例 ，下面我们来学习一下 rabbitMQ 的消息应答 与 发布
1. 消息应答
引用:
消费者完成一个任务可能需要一段时间，如果其中一个消费者处理一个长的任务并仅只完成了部分突然它挂掉了，会发生什么情况。RabbitMQ 一旦向消费者传递了一条消息，便立即将该消息标记为删除。在这种情况下，突然有个消费者挂掉了，我们将丢失正在处理的消息。以及后续发送给该消费者的消息，因为它无法接收到。
为了保证消息在发送过程中不丢失，引入消息应答机制 (又称 确认机制)，消息应答就是：消费者在接收到消息并且处理该消息之后，告诉 rabbitmq 它已经处理了，rabbitmq 可以把该消息删除了。
这个就当前与在 网络中学习到的 应答机制. 主机A 发送一个数据包给 主机B ， 主机A 就会等待 主机B 发送一个 ACK 回来 如果 主机A 收到了 主机B 的 ack 说明发送成功 ，如果没有收到就说明 发送失败 重新发送 .
关于 消息应答 rabbitmq 提供两种模式
- 自动应答
- 手动应答
这里先来看看 自动应答
1.2 自动应答
概念 : 在自动应答模式下，消费者从队列中获取消息后，RabbitMQ 会自动将消息标记为已传递（Delivered）状态，不需要消费者明确发送应答给 RabbitMQ。这种模式下，RabbitMQ 会立即将消息从队列中删除，并假设消息已经被成功处理。
需要注意: 自动应答需要良好的环境 , 不能存在极端的情况，简单来说 自动应答 不是很靠谱
这种模式需要在高吞吐量和数据传输安全性方面做权衡,因为这种模式如果消息在接收到之前，消费者那边出现连接或者 channel 关闭，那么消息就丢失了,当然另一方面这种模式消费者那边可以传递过载的消息，没有对传递的消息数量进行限制，当然这样有可能使得消费者这边由于接收太多还来不及处理的消息，导致这些消息的积压，最终使得内存耗尽，最终这些消费者线程被操作系统杀死，**所以这种模式仅适用在消费者可以高效并以 某种速率能够处理这些消息的情况下使用
自动应答简单看看，下面我们来学习一下手动应答
1.2 手动应答
手动应答概念 :
在默认情况下，消费者从队列中获取消息后，RabbitMQ 会立即将该消息标记为已传递（Delivered）状态。然而，在手动应答模式下，消费者需要明确告知 RabbitMQ 消息是否已被消费并处理完成。如果消费者成功处理了消息，则发送一个应答给 RabbitMQ，RabbitMQ 将该消息从队列中删除；如果消费者未发送应答或应答失败，则 RabbitMQ 认为消息未被成功处理，会将其重新发送给其他消费者。
手动消息应答使用的方法
Channel.basicAck
： 手动应答消息的方法
basicAck(long deliveryTag,boolean multiple);1. deliveryTag: 表示消息的唯一标识符。每条消息都会被分配一个唯一的 deliveryTag。消费者在处理完一条消息后，需要将对应的 deliveryTag 传递给 basicAck 方法，以告知 RabbitMQ 哪条消息已经被成功处理。 --> 简单来说就是消息的标记
2. multiple: 表示是否批量确认。如果设置为 true，则表示除了确认当前指定的 deliveryTag 对应的消息外，还要确认所有之前未确认的消息；如果设置为 false，则只确认当前指定的 deliveryTag 对应的消息。 --> 简单来说就是 rabbitmq 确定了当前的消息,可以将其丢弃了
Channel.basicReject
: 拒绝 一条消息的方法
basicReject(long deliveryTag,boolean requeue);1. deliveryTag：表示消息的唯一标识符，与 basicAck 方法中的参数相同。每条消息都会被分配一个唯一的 deliveryTag。
2. 表示是否重新将消息放回队列中进行重新投递。如果设置为 false，则消息会被直接丢弃；如果设置为 true，则消息会被重新放回队列，等待被消费者重新处理
Channel.basicNack
: 用于一次性拒绝多个消息。
basicNack(long deliveryTag,boolean multiple,boolean requeue);
这里 deliveryTag 和 requeue 和 Channel.basicReject 一样的， 这个方法只是多了一个 basicNack 方法用来 表示拒绝的多个参数
multiple: 表示是否拒绝所有 deliveryTag 小于等于当前 deliveryTag 的消息。如果设置为 false，则只拒绝当前 deliveryTag 对应的消息；如果设置为 true，则会拒绝所有 deliveryTag 小于等于当前 deliveryTag 的消息。
Channel.basicRecover
: 用于重新消费未被确认的方法
basicRecover(boolean requeue);
requeue: 表示是否将未被确认的消息重新放回队列中等待投递。如果设置为 false，则未被确认的消息会被删除，否则它们将被重新排队并可供之后的消费者使用。
在上面这些方法中 又 multiple 参数 这里简单讲讲一下 multiple.
手动应答的好处是可以批量应答并且介绍网络阻塞，手动应答就是通过 multiple 参数来开启是否批量
multiple 取 true 表示 批量处理 channel （信道）中为应答的消息
multiple 取 false 就不会开启批量处理的功能
看完 multiple ，再来谈谈 重新消费，当有些消息被拒绝后会被重新放到队列中 ，等待被重新消费， 除了这种方法, 还有一种情况会导致消息重新入队 。
这种方式就是: 如果消费者由于某些原因失去连接(其通道已关闭，连接已关闭或 TCP 连接丢失)，导致消息未发送 ACK 确认，RabbitMQ 将了解到消息未完全处理，并将对其重新排队。
如果此时其他消费者可以处理，它将很快将其重新分发给另一个消费者。这样，即使某个消费者偶尔死亡，也可以确保不会丢失任何消息。
图: 消费者 c1 出现问题，未返回 ack ，消息重新回到队列 ，被其他消费者（c2）消费.
看完概念，就来写写代码 ， 体验体验.
需要注意: rabbimt 默认采取的是 自定应答，要实现消息在消费过程中不被丢失，需要手动开启手动应答
1.3 代码案例
这里 创建一个消费者 ，消费者开启两个线程 ， 消费者1 一秒消费 一个消息 ， 消费者 20秒消费一个消息，然后在 消费者2 消费消息的时候，停止运行 (假设消费者2 宕机了) ， 然后查看 消费者1 是否消费了 消费者2 没有消费的消息 (验证消息是否回到 队列，并重新安排消费者进行消费)
生产者:
packageorg.example.trhree;importcom.rabbitmq.client.Channel;importorg.example.utils.RabbitMQUtils;importjava.io.IOException;importjava.util.Scanner;importjava.util.concurrent.TimeoutException;/**
* 消息再手动应答时不丢失,放回队列重新消费
*/publicclassTask{// 队列名称publicstaticfinalStringTASK_QUEUE_NAME="ack_queue";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Channel channel =RabbitMQUtils.getChannel();// 声明一个队列
channel.queueDeclare(TASK_QUEUE_NAME,false,false,false,null);// 从控制台中输入信息Scanner sc =newScanner(System.in);while(sc.hasNext()){String message = sc.next();
channel.basicPublish("",TASK_QUEUE_NAME,null, message.getBytes("UTF-8"));System.out.println("消息发送成功: "+ message);}}}
消费者1
packageorg.example.trhree;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.DeliverCallback;importorg.example.utils.RabbitMQUtils;importjava.io.IOException;importjava.util.concurrent.TimeoutException;/**
* 消费在手动应答时不丢失,放回到队列重新消费
*/publicclassWork1{// 队列名称publicstaticfinalStringTASK_QUEUE_NAME="ack_queue";//接收消息publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Channel channel =RabbitMQUtils.getChannel();System.out.println("C1 等待接受消息处理事件较短");DeliverCallback deliverCallback =(tag, message)->{// 沉睡一秒try{Thread.sleep(1000);}catch(InterruptedException e){thrownewRuntimeException(e);}System.out.println("c1 接受到的消息: "+newString(message.getBody(),"UTF-8"));// 手动应答/**
* basicAck 参数
* 1. 消息的标记
* 2. 是否批量应答 : true 批量 false 不批量
*/
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);};// 采用手动应答boolean autoAck =false;
channel.basicConsume(TASK_QUEUE_NAME, autoAck, deliverCallback,(tag ->{System.out.println("消费者取消消费: "+ tag);}));}}
消费者2 ： 代码和消费者1 一毛一样 ， 改改 log 里面打印即可
packageorg.example.trhree;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.DeliverCallback;importorg.example.utils.RabbitMQUtils;importjava.io.IOException;importjava.util.concurrent.TimeoutException;/**
* 消费在手动应答时不丢失,放回到队列重新消费
*/publicclassWork2{// 队列名称publicstaticfinalStringTASK_QUEUE_NAME="ack_queue";//接收消息publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Channel channel =RabbitMQUtils.getChannel();System.out.println("C2 等待接受消息处理事件较长");DeliverCallback deliverCallback =(tag, message)->{// 沉睡20秒 &&& 注意消费者是要消费 20 秒的try{Thread.sleep(20000);}catch(InterruptedException e){thrownewRuntimeException(e);}System.out.println("c1 接受到的消息: "+newString(message.getBody(),"UTF-8"));// 手动应答/**
* basicAck 参数
* 1. 消息的标记
* 2. 是否批量应答 : true 批量 false 不批量
*/
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);};// 采用手动应答boolean autoAck =false;
channel.basicConsume(TASK_QUEUE_NAME, autoAck, deliverCallback,(tag ->{System.out.println("消费者取消消费: "+ tag);}));}}
效果:
2. RabbitMQ 持久化
默认情况下，RabbitMQ 创建的队列是非持久化的，这意味着在 RabbitMQ 服务器停止或重启时，队列将被删除。
我们想要在 RabbitMQ 服务暂停或重启以后 ，消息的生产者发送过来的消息不丢失，就需要来学习一下 如何开启 rabbitmq 持久化 .
2.1 队列持久化
在之前的例子中，我们所创建的队列都是非持久化的， RabbitMQ 重启或者故障 导致 宕机 ,队列都会被删除 这肯定是不行的， 这里我们想要队列持久化 就要 在声明队列的时候 将 durable 属性值设置为 true .
packageorg.example.trhree;importcom.rabbitmq.client.Channel;importorg.example.utils.RabbitMQUtils;importjava.io.IOException;importjava.util.Scanner;importjava.util.concurrent.TimeoutException;publicclassTask2{// 队列名称publicstaticfinalStringTASK_QUEUE_NAME="ack_queue";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Channel channel =RabbitMQUtils.getChannel();// 开启持久化boolean durable =true;// 声明一个队列 --> 将第二个参数设置为 ture 就表示这个队列 是持久化的.
channel.queueDeclare(TASK_QUEUE_NAME, durable,false,false,null);// 从控制台中输入信息Scanner sc =newScanner(System.in);while(sc.hasNext()){String message = sc.next();
channel.basicPublish("",TASK_QUEUE_NAME,null, message.getBytes("UTF-8"));System.out.println("消息发送成功: "+ message);}}}
声明完后 ack_queue 队列就是持久化的， 但是需要注意一点 ，如果 ack_queue 之前 声明的时候不是持久化的 (再次声明会报错)，就需要先把 ack_queue 队列删除掉 或者重新建立一个 持久化的队列.
这里 通过上面的代码声明出来的 ack_queue 队列 在 Features 就会多出 一个 D ，表示 这是一个持久化的队列。
2.2 消息持久化
我们将队列设置为持久化 ，默认情况下 消息是仍然是非持久化的 （rabbitmq 服务器段宕机后 队列还在 ，但是队列里的消息没了） .
这里我们想要消息持久化 需要在
basicPublish
方法的第二个参数添加这个 属性:
MessageProperties.PERSISTENT_TEXT_PLAIN
MessageProperties 类型
引用:
BASIC: 默认的消息属性，无特殊配置。MINIMAL_BASIC: 最小化的基本属性，不包含任何附加属性。PERSISTENT_BASIC: 基本持久化属性，将消息设置为持久化。PERSISTENT_TEXT_PLAIN: 文本类型的持久化属性，将消息设置为持久化，并指定内容类型为文本。PERSISTENT_BINARY: 二进制类型的持久化属性，将消息设置为持久化，并指定内容类型为二进制。NON_PERSISTENT_BASIC: 基本非持久化属性，将消息设置为非持久化。
packageorg.example.trhree;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.MessageProperties;importorg.example.utils.RabbitMQUtils;importjava.io.IOException;importjava.util.Scanner;importjava.util.concurrent.TimeoutException;publicclassTask2{// 队列名称publicstaticfinalStringTASK_QUEUE_NAME="ack_queue";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Channel channel =RabbitMQUtils.getChannel();// 开启持久化boolean durable =true;// 声明一个队列
channel.queueDeclare(TASK_QUEUE_NAME, durable,false,false,null);// 从控制台中输入信息Scanner sc =newScanner(System.in);while(sc.hasNext()){String message = sc.next();// 通过 MessageProperties.PERSISTENT_TEXT_PLAIN 开启消息持久化
channel.basicPublish("",TASK_QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));System.out.println("消息发送成功: "+ message);}}}
这里虽然 设置了 消息 持久化 ，但是消息并不能完全保证不丢失, 尽管它告诉 RabbitMQ 将消息保存到磁盘，但是这里依然存在当消息刚准备存储在磁盘的时候 但是还没有存储完，消息还在缓存的一个间隔点。此时并没 有真正写入磁盘。持久性保证并不强，但是对于我们的简单任务队列而言，这已经绰绰有余了。
3. 不公平分发
看完 rabbitmq 的 消息应答 和 持久化 ，下面我们来说说 不公平分发 ,在之前的文章说过 rabbitmq 采用 轮询的方式来分发消息 , 但是在某种场景下 轮询分发 并不是很好的选择， 比如 上面消息持久化举得例子 ，存在两个消费者 ， 消费者1 消费消息 需要 1 秒钟 ， 消费者2 消费消息 需要 10 分钟 ，采用轮询的方法 就会导致 消费者1 一直处于 空闲状态 ，而 消费者 2 一直处于工作状态 ，明显是不好的 （消费者1 处理完消息 一直等待 消费者2 处理完消息 ） ， 按照 常理来说 ，我们应该给 有能力者 安排多一点的工作 ，能力比较差的 ，少安排一点工作 .
rabbitmq 就考虑到了这种情况 ，当 有一方 处理 能力比较 低 ，一方处理能力比较高 ，就会 给 较高的一方 安排多一点 任务 ， 低的一方 安排少一点 .
我们想要 使用 不公平分发 这种 模式 需要 在 消费者消费消息之前 ，设置 参数
channel.basicQos(1)
(轮询是 basicQos 设置为 0)
代码演示:
packageorg.example.trhree;importcom.rabbitmq.client.CancelCallback;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.DeliverCallback;importorg.example.utils.RabbitMQUtils;importjava.io.IOException;importjava.util.concurrent.TimeoutException;publicclassWork3{// 队列名称publicstaticfinalStringTASK_QUEUE_NAME="ack_queue";// 接受消息publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Channel channel =RabbitMQUtils.getChannel();DeliverCallback deliverCallback =(tag, message)->{System.out.println("手动应答处理消息");// 手动应答
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);};CancelCallback cancelCallback =(tag ->{System.out.println(tag +"消费者取消消费接口回调逻辑");});// 设置不公平分发int prefetchCount =1;
channel.basicQos(prefetchCount);// 采用手动应答boolean autoAck =false;
channel.basicConsume(TASK_QUEUE_NAME, autoAck, deliverCallback, cancelCallback);}}
work4 : 拷贝 work3 ， 在 处理消息的回调 deliver中 睡眠 10 ，
packageorg.example.trhree;importcom.rabbitmq.client.CancelCallback;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.DeliverCallback;importorg.example.utils.RabbitMQUtils;importjava.io.IOException;importjava.util.concurrent.TimeoutException;publicclassWork4{// 队列名称publicstaticfinalStringTASK_QUEUE_NAME="ack_queue";// 接受消息publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Channel channel =RabbitMQUtils.getChannel();DeliverCallback deliverCallback =(tag, message)->{System.out.println("手动应答处理消息");try{Thread.sleep(10000);}catch(InterruptedException e){thrownewRuntimeException(e);}// 手动应答
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);};CancelCallback cancelCallback =(tag ->{System.out.println(tag +"消费者取消消费接口回调逻辑");});// 设置不公平分发int prefetchCount =1;
channel.basicQos(prefetchCount);// 采用手动应答boolean autoAck =false;
channel.basicConsume(TASK_QUEUE_NAME, autoAck, deliverCallback, cancelCallback);}}
启动一下看看效果:
可以看到 ， 此时 work3 就处理了多条消息 .
最后在说说 不公平分发思想:
不公平分发思想：如果一个工作队列还没有处理完或者没有应答签收一个消息，则不拒绝 RabbitMQ 分配新的消息到该工作队列。此时 RabbitMQ 会优先分配给其他已经处理完消息或者空闲的工作队列。如果所有的消费者都没有完成手上任务，队列还在不停的添加新任务，队列有可能就会遇到队列被撑满的情况，这个时候就只能添加新的 worker (工作队列)或者改变其他存储任务的策略。
看完不公平分发，我们来看看预取值分发 。
4. 预取值分发
引用:
预取值分发是带权的消息分发 , 默认消息的发送是异步发送的，所以在任何时候，channel 上不止只有一个消息来自消费者的手动确认，所以本质上是异步的。因此这里就存在一个未确认的消息缓冲区，希望开发人员能限制此缓冲区的大小，以避免缓冲区里面无限制的未确认消息问题。这个时候就可以通过使用
basic.basicQos方法设置「预取计数」值来完成的。
该值定义通道上允许的未确认消息的最大数量。一旦数量达到配置的数量， RabbitMQ 将停止在通道上传递更多消息，除非至少有一个未处理的消息被确认，例如，假设在通道上有未确认的消息 5、6、7，8，并且通道的预取计数设置为 4，此时 RabbitMQ 将不会在该通道上再传递任何消息，除非至少有一个未应答的消息被 ack。比方说 tag=6 这个消息刚刚被确认 ACK，RabbitMQ 将会感知这个情况到并再发送一条消息。消息应答和 QoS 预取值对用户吞吐量有重大影响。
通常，增加预取将提高向消费者传递消息的速度。虽然自动应答传输消息速率是最佳的，但是，在这种情况下已传递但尚未处理的消息的数量也会增加，从而增加了消费者的 RAM 消耗(随机存取存储器)应该小心使用具有无限预处理的自动确认模式或手动确认模式，消费者消费了大量的消息如果没有确认的话，会导致消费者连接节点的内存消耗变大，所以找到合适的预取值是一个反复试验的过程，不同的负载该值取值也不同 100 到 300 范围内的值通常可提供最佳的吞吐量，并且不会给消费者带来太大的风险。
预取值为 1 是最保守的。当然这将使吞吐量变得很低，特别是消费者连接延迟很严重的情况下，特别是在消费者连接等待时间较长的环境 中。对于大多数应用来说，稍微高一点的值将是最佳的。
图解:
代码演示:
生产者:
packageorg.example.four;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.MessageProperties;importorg.example.utils.RabbitMQUtils;importjava.io.IOException;importjava.util.Scanner;importjava.util.concurrent.TimeoutException;publicclassTask{// 队列名称publicstaticfinalStringTASK_QUEUE_NAME="test_queue_expect";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Channel channel =RabbitMQUtils.getChannel();// 开启持久化boolean durable =true;// 声明一个队列
channel.queueDeclare(TASK_QUEUE_NAME, durable,false,false,null);// 从控制台中输入信息Scanner sc =newScanner(System.in);while(sc.hasNext()){String message = sc.next();// 通过 MessageProperties.PERSISTENT_TEXT_PLAIN 开启消息持久化
channel.basicPublish("",TASK_QUEUE_NAME,MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8"));System.out.println("消息发送成功: "+ message);}}}
消费者 c1
packageorg.example.four;importcom.rabbitmq.client.CancelCallback;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.DeliverCallback;importorg.example.utils.RabbitMQUtils;importjava.io.IOException;importjava.util.concurrent.TimeoutException;// 消费者 c1publicclassWorkC1{// 队列名称publicstaticfinalStringTASK_QUEUE_NAME="test_queue_expect";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Channel channel =RabbitMQUtils.getChannel();DeliverCallback deliverCallback =(tag, message)->{System.out.println("消费者 c1 接收到消息: "+newString(message.getBody()));try{Thread.sleep(1000);}catch(InterruptedException e){thrownewRuntimeException(e);}
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);};CancelCallback cancelCallback =(tag ->{System.out.println(tag +"消费者取消消费接口回调逻辑");});// 设置 c1 的预期取值为 2int prefetchCount =2;
channel.basicQos(prefetchCount);// 采用 手动应答boolean autoAck =false;
channel.basicConsume(TASK_QUEUE_NAME, autoAck, deliverCallback, cancelCallback);}}
消费者 c2 :
packageorg.example.four;importcom.rabbitmq.client.CancelCallback;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.DeliverCallback;importorg.example.utils.RabbitMQUtils;importjava.io.IOException;importjava.util.concurrent.TimeoutException;// 消费者 c1publicclassWorkC2{// 队列名称publicstaticfinalStringTASK_QUEUE_NAME="test_queue_expect";publicstaticvoidmain(String[] args)throwsIOException,TimeoutException{Channel channel =RabbitMQUtils.getChannel();DeliverCallback deliverCallback =(tag, message)->{System.out.println("消费者 c2 接收到消息: "+newString(message.getBody()));try{Thread.sleep(10000);}catch(InterruptedException e){thrownewRuntimeException(e);}
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);};CancelCallback cancelCallback =(tag ->{System.out.println(tag +"消费者取消消费接口回调逻辑");});// 设置 c2 的预期取值为 5int prefetchCount =5;
channel.basicQos(prefetchCount);// 采用 手动应答boolean autoAck =false;
channel.basicConsume(TASK_QUEUE_NAME, autoAck, deliverCallback, cancelCallback);}}
效果:
总共发送了 10 条消息 ， c2 处理一条消息需要10秒钟 ，c1处理一条消息 需要 1 秒钟， 可以看到上面的图中 ，c1 消费了 5 条消息 ， 应为 c2 预期接受到的消息是 5 条 ，所以 ，c2 消费完 预取值的 2 条消息 ，rabbitmq 就会 再发 2 条消息，又因为 c2 处能力 比较弱 ，还没有处理完 预取值的消息 ， rabbitmq 就将 最后一条消息 交给了 c1 处理.
最终:
到此 预取值分发 看完， 我们应该能发现 ， 使用 不公平分发或预取值分发， 都是 使用
basicQos
方法 .
当 basicQos 方法 取值为 1 的时候是 不公平分发 ， basicQos 取其他整数值时为 预取值分发 (取 0 为 轮询分发).
两种分发方式看完，下面我们来学习 发布确认 (发布确认是保证消息不丢失的重要环节) .
5. 发布确认
生产者发布消息到 RabbitMQ 后，需要 RabbitMQ 返回「ACK（已收到）」给生产者，这样生产者才知道自己生产的消息成功发布出去。
5.1 发布确认逻辑
引用:
生产者将信道设置成 confirm 模式，一旦信道进入 confirm 模式，所有在该信道上面发布的消息都将会被指派一个唯一的 ID(从 1 开始)，一旦消息被投递到所有匹配的队列之后，broker 就会发送一个确认给生产者(包含消息的唯一 ID)，这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker 回传给生产者的确认消息中 delivery-tag 域包含了确认消息的序列号，此外 broker 也可以设置 basic.ack 的 multiple 域，表示到这个序列号之前的所有消息都已经得到了处理。
confirm 模式最大的好处在于是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果RabbitMQ 因为自身内部错误导致消息丢失，就会发送一条 nack 消息， 生产者应用程序同样可以在回调方法中处理该 nack 消息。
5.2 开启发布确认的方法
发布确认默认是没有开启的，如果要开启需要调用方法
confirmSelect
，每当你要想使用发布确认，都需要在
channel
上调用该方法
//开启发布确认
channel.confirmSelect();// 确认消息 (持久化完成)
channel.waitForConfirms();
在确认发布中 有三种 方法 ：
- 单独确认
- 批量确认
- 异步确认
接下来我们一个一个学习 . 另外我会在 代码案例中 通过计算确认时间 来展示三种确认的发布速度.
5.3 单个确认发布
这是一种简单的确认方式，它是一种同步确认发布的方式，也就是发布一个消息之后只有它被确认发布，后续的消息才能继续发布，
waitForConfirmsOrDie(long)
这个方法只有在消息被确认的时候才返回，如果在指定时间范围内这个消息没有被确认那么它将抛出异常。
这种确认方式有一个最大的缺点就是：发布速度特别的慢，因为如果没有确认发布的消息就会阻塞所有后续消息的发布，这种方式最多提供每秒不超过数百条发布消息的吞吐量。当然对于某些应用程序来说这可能已经足够了。
代码案例:
packageorg.example.four.tow;importcom.rabbitmq.client.Channel;importorg.example.utils.RabbitMQUtils;importjava.io.IOException;importjava.util.UUID;importjava.util.concurrent.TimeoutException;publicclassConfirmMessage{// 单个发消息的个数publicstaticfinalintMESSAGE_COUNT=1000;publicstaticvoidmain(String[] args)throwsIOException,TimeoutException,InterruptedException{publishMessageIndividually();// 发布 1000 个消息 并确认 耗时:}// 单个确认publicstaticvoidpublishMessageIndividually()throwsIOException,TimeoutException,InterruptedException{Channel channel =RabbitMQUtils.getChannel();// 队列的声明String queueName =UUID.randomUUID().toString();
channel.queueDeclare(queueName,false,true,false,null);// 开去发布确认
channel.confirmSelect();// 开始时间long begin =System.currentTimeMillis();for(int i =0; i <MESSAGE_COUNT; i++){String message = i +"";
channel.basicPublish("", queueName,null, message.getBytes());// 单个消息 --> 通过 waitForConfirms 确认这个消息boolean flag = channel.waitForConfirms();if(flag){System.out.println("消息发布成功!");}}// 结束时间long end =System.currentTimeMillis();System.out.println("发布"+MESSAGE_COUNT+"个消息,每个消息单独确认, 耗时: "+(end - begin)+"ms");}}
效果：
单个 看完 下面我们来看 批量确认发布
5.4 批量确认发布
引用:
单个确认发布方式非常慢，与单个等待确认消息相比，先发布一批消息然后一起确认可以极大地提高吞吐量，当然这种方式的缺点就是：当发生故障导致发布出现问题时，不知道是哪个消息出问题了，我们必须将整个批处理保存在内存中，以记录重要的信息而后重新发布消息。当然这种方案仍然是同步的，也一样阻塞消息的发布。
代码案例:
packageorg.example.four.tow;importcom.rabbitmq.client.Channel;importorg.example.utils.RabbitMQUtils;importjava.io.IOException;importjava.util.UUID;importjava.util.concurrent.TimeoutException;publicclassConfirmMessage2{// 发送消息的个数publicstaticfinalintMESSAGE_COUNT=1000;publicstaticvoidmain(String[] args)throwsIOException,TimeoutException,InterruptedException{publishMessageIndividually();// 发布 1000 个消息 并确认 耗时:}// 批量发布确认publicstaticvoidpublishMessageIndividually()throwsIOException,TimeoutException,InterruptedException{Channel channel =RabbitMQUtils.getChannel();// 队列的声明String queueName =UUID.randomUUID().toString();
channel.queueDeclare(queueName,false,true,false,null);// 开去发布确认
channel.confirmSelect();// 开始时间long begin =System.currentTimeMillis();// 批量确认消息大小int batchSize =100;for(int i =0; i <MESSAGE_COUNT; i++){String message = i +"";
channel.basicPublish("", queueName,null, message.getBytes());// 单个消息 --> 通过 waitForConfirms 确认这个消息if((i +1)% batchSize ==0){// 此时 发送消息 到了 100 , 使用 waitForConfirms 确认消息
channel.waitForConfirms();}}// 结束时间long end =System.currentTimeMillis();System.out.println("发布"+MESSAGE_COUNT+"个消息,每次批量确认100个消息, 耗时: "+(end - begin)+"ms");}}
效果:
对比一下 单个确认发布 耗费的时间 ，批量确认 ， 明显快很多 （上面 单个确认 ，有一个 打印 所以时间会比 批量确认耗费时间很多）
最后我们来看一下 异步确认
5.5 异步确认
异步确认 会比 批量确认 和 单个确认 在编程逻辑上 复杂很多 ，但也带来了性能上的优势 ， 异步确认 效率和可靠性都非常好，是 通过 回调函数来达到消息可靠传递的.
上面这张图 大致 流程是 ， 消息生产者 发送消息 发送给 broker ，broker 会通过 ackCallback 回调函数 告诉 生产者 那些 消息是成功应答了的 ，失败的消息 会通过 nackCallback 告诉生产者 ，
代码案例:
packageorg.example.four.tow;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.ConfirmCallback;importorg.example.utils.RabbitMQUtils;importjava.io.IOException;importjava.util.UUID;importjava.util.concurrent.TimeoutException;publicclassConfirmMessage3{// 发送消息的个数publicstaticfinalintMESSAGE_COUNT=1000;publicstaticvoidmain(String[] args)throwsIOException,TimeoutException,InterruptedException{publishMessageIndividually();// 发布 1000 个消息 并确认 耗时:}// 批量发布确认publicstaticvoidpublishMessageIndividually()throwsIOException,TimeoutException,InterruptedException{Channel channel =RabbitMQUtils.getChannel();// 队列的声明String queueName =UUID.randomUUID().toString();
channel.queueDeclare(queueName,false,true,false,null);// 开去发布确认
channel.confirmSelect();// 开始时间long begin =System.currentTimeMillis();/**
* 1. 消息的标记
* 2. 是否批量确认
*/// 消息确认 回调的函数ConfirmCallback ackCallback =(deliveryTag, multiple)->{System.out.println("确认的消息： "+ deliveryTag);};ConfirmCallback nackCallback =(deliveryTag, multiple)->{System.out.println("未确认的消息： "+ deliveryTag);};// 消息的监听器 ,监听那些消息成功了 ,那些消息失败了
channel.addConfirmListener(ackCallback, nackCallback);// 异步通知/**
* 参数解释
* 1. 监听那些消息成功了
* 2. 监听那些消息失败了
*/// 发送消息for(int i =0; i <MESSAGE_COUNT; i++){String message = i +"";
channel.basicPublish("", queueName,null, message.getBytes());}// 结束时间long end =System.currentTimeMillis();System.out.println("发布"+MESSAGE_COUNT+"个消息,通过异步确认,进行确认耗时: "+(end - begin)+"ms");}}
效果: 这里 最后 发布 xxxx 消息 在 确认消息 之前 是因为 监听 消息成功 和 失败的 回调函数 是 异步的.
可以看到 时间是 28ms ， 注意这里是带 打印语句了 是会 耗费一点时间的 ， 很明显这是 比 批量确认 和 单个确认消息 效率高的.
看完了代码案例， 再来讲一个很重要的 ，就是 如何处理异步未确认消息 , 上面的 案例 打印出来的 全是 已确认的 ，在以后面对的场景中肯定会出现 ，消息未确认的情况 ，那么 我们要如何 消费者知道 未确认的消息并 重新发送呢?
要解决这个问题 下面我们就来学习一下 处理异步未确认消息
5.5.1 处理异步未确认消息
关于异步处理未确认的消息: 最好的解决的解决方案就是把未确认的消息放到一个基于内存的能被发布线程访问的队列，比如说用 ConcurrentLinkedQueue 这个队列在 confirm ， callbacks 与发布线程之间进行消息的传递。
代码案例：
packageorg.example.four.tow;importcom.rabbitmq.client.Channel;importcom.rabbitmq.client.ConfirmCallback;importorg.example.utils.RabbitMQUtils;importjava.io.IOException;importjava.util.UUID;importjava.util.concurrent.ConcurrentNavigableMap;importjava.util.concurrent.ConcurrentSkipListMap;importjava.util.concurrent.TimeoutException;publicclassConfirmMessage4{// 发送消息的个数publicstaticfinalintMESSAGE_COUNT=1000;publicstaticvoidmain(String[] args)throwsIOException,TimeoutException,InterruptedException{publishMessageIndividually();// 发布 1000 个消息 并确认 耗时:}// 批量发布确认publicstaticvoidpublishMessageIndividually()throwsIOException,TimeoutException,InterruptedException{// 准备一个 线程安全有序的一个哈希表 适用于高并发的情况下// 1. 轻松的将序号与消息进行关联// 2. 轻松批量删除条目 只要给到序号// 3. 支持高并发 （多线程）ConcurrentSkipListMap<Long,String> outStandingConfirms =newConcurrentSkipListMap<>();Channel channel =RabbitMQUtils.getChannel();// 队列的声明String queueName =UUID.randomUUID().toString();
channel.queueDeclare(queueName,false,true,false,null);// 开去发布确认
channel.confirmSelect();// 开始时间long begin =System.currentTimeMillis();/**
* 1. 消息的标记
* 2. 是否批量确认
*/// 消息确认 回调的函数ConfirmCallback ackCallback =(deliveryTag, multiple)->{if(multiple){// 删除所有已经确认的消息 ,剩下的就是未确认的消ConcurrentNavigableMap<Long,String> confirmed = outStandingConfirms.headMap(deliveryTag);/**
* headMap 方法用于将已确认的消息存入新的 Map 缓存区里 （标记小于 deliveryTag的信息），
* 然后手动清除该新缓存区的内容。因为 headMap 方法是浅拷贝，所以清除了缓存区，相当于清除了内容的地址，
* 也就清除了队列的确认的消息。
*/}else{// 不是 批量确认 , 通过 remove 删除确认的消息
outStandingConfirms.remove(deliveryTag);}System.out.println("确认的消息： "+ deliveryTag);};ConfirmCallback nackCallback =(deliveryTag, multiple)->{// 处理未确认的消息String message = outStandingConfirms.get(deliveryTag);System.out.println("未确认的消息： "+message+" 编号为: "+ deliveryTag);};// 消息的监听器 ,监听那些消息成功了 ,那些消息失败了
channel.addConfirmListener(ackCallback, nackCallback);// 异步通知/**
* 参数解释
* 1. 监听那些消息成功了
* 2. 监听那些消息失败了
*/// 发送消息for(int i =0; i <MESSAGE_COUNT; i++){String message = i +"";
channel.basicPublish("", queueName,null, message.getBytes());// 记录所有要发送的消息
outStandingConfirms.put(channel.getNextPublishSeqNo(), message);}// 结束时间long end =System.currentTimeMillis();System.out.println("发布"+MESSAGE_COUNT+"个消息,通过异步确认,进行确认耗时: "+(end - begin)+"ms");}}
效果:
最后对 三种发布确认消息速度对比:
- 单独发布消息同步等待确认，简单，但吞吐量非常有限。
- 批量发布消息批量同步等待确认，简单，合理的吞吐量，一旦出现问题但很难推断出是那条消息出现了问题。
- 异步处理最佳性能和资源使用，在出现错误的情况下可以很好地控制，但是实现起来稍微难些
最最后面，在来说一下应答和发布的区别 ， 应答功能属于消费者，当消费者消费完消息后告诉 rabbitmq 消费成功 ，发布属于生产者，生产者生产的消息到达 rabbitmq 后 ，rabbitmq 告诉生产者接收到消息.
版权归原作者 牧.. 所有， 如有侵权，请联系我们删除。