消息应答

概念

自动应答

消息应答的方法

Multiple 的解释

手动应答实现

1.准备工具类

2.生产者

3.两个睡眠时间不同的消费者

4.效果展示

消息自动重新入队

效果演示：

消息应答

概念

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

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

自动应答

消息发送后立即被认为已经传送成功

弊端：如果消息在接收到之前，消费者那边出现连接或者信道关闭，那么消息就丢失了,当然另一方面这种模式消费者那边可以传递过载的消息，没有对传递的消息数量进行限制，当然这样有可能使得消费者这边由于接收太多还来不及处理的消息，导致这些消息的积压，最终使得内存耗尽，最终这些消费者线程被操作系统杀死

所以在实际开发中我们应选择手动应答

消息应答的方法

Channel.basicAck()：用于肯定确认

RabbitMQ 已知道该消息并且成功的处理消息，可以将其丢弃了

Channel.basicNack()：用于否定确认

Channel.basicReject()：用于否定确认（推荐使用）

Multiple 的解释

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

multiple 的 true 和 false 代表不同意思

true 代表批量应答 channel 上未应答的消息

比如说 channel 上有传送 tag 的消息 5,6,7,8 当前 tag 是 8

那么此时 5-8 的这些还未应答的消息都会被确认收到消息应答

false 同上面相比

只会应答 tag=8 的消息 5,6,7 这三个消息依然不会被确认收到消息应答

手动应答实现

默认消息采用的是自动应答，所以我们要想实现消息消费过程中不丢失，需要把自动应答改为手动应答

1.准备工具类

/*** 连接工厂创建信道的工具类*/
public class RabbitMQUtil {public static Channel getChannel() throws Exception {//创建一个连接工厂ConnectionFactory factory = new ConnectionFactory();//设置工厂的IP，连接RabbitMQ的队列factory.setHost("192.168.31.65");//设置用户名factory.setUsername("admin");//密码factory.setPassword("123");//创建连接Connection connection = factory.newConnection();//获取信道return connection.createChannel();}
}

/*** 睡眠：模拟执行业务逻辑操作需要时间*/
public class SleepUtil {public static void sleep(int second) {try {Thread.sleep(1000 * second);} catch (InterruptedException _ignored) {Thread.currentThread().interrupt();}}
}

2.生产者

/*** 消息在手动应答时不丢失，放回队列后重新消费*/
public class Producer {public static final String TASK_QUEUE_NAME = "ack_queue";public static void main(String[] args) throws Exception {Channel channel = RabbitMQUtil.getChannel();//开启发布确认channel.confirmSelect();//声明队列channel.queueDeclare(TASK_QUEUE_NAME, false, false, false, null);//从控制台中输入信息Scanner scanner = new Scanner(System.in);while (scanner.hasNext()) {String message = scanner.next();channel.basicPublish("", TASK_QUEUE_NAME, null, message.getBytes("UTF-8"));System.out.println("生产者发出消息" + message);}}
}

3.两个睡眠时间不同的消费者

/*** 消息在手动应答时不丢失，放回队列后重新消费*/
public class Worker01 {//队列的名称public static final String TASK_QUEUE_NAME = "ack_queue";//接受消息public static void main(String[] args) throws Exception {//获取信道Channel channel = RabbitMQUtil.getChannel();System.out.println("消费者一：等待接受___时间较短");DeliverCallback deliverCallback = (consumerTag, message) -> {//沉睡1秒，模拟执行代码需要一定的时间SleepUtil.sleep(1);System.out.println("接受到的消息" + new String(message.getBody()));//进行手动应答/** 1.消息的标记 tag，表明消息的唯一标识* 2.是否批量应答 应该不允许批量，防止消息的丢失* */channel.basicAck(message.getEnvelope().getDeliveryTag(), false);};//消息被取消时执行下面的内容CancelCallback cancelCallback = consumerTag -> {System.out.println(consumerTag + "消息消费被中断");};//消费者进行消息的消费channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, cancelCallback); //采用手动应答（false）}
}

/*** 消息在手动应答时不丢失，放回队列后重新消费*/
public class Worker02 {//队列的名称public static final String TASK_QUEUE_NAME = "ack_queue";//接受消息public static void main(String[] args) throws Exception {//获取信道Channel channel = RabbitMQUtil.getChannel();System.out.println("消费者二：等待接受___时间较长");DeliverCallback deliverCallback = (consumerTag, message) -> {//沉睡10秒，模拟执行代码需要一定的时间SleepUtil.sleep(10);System.out.println("接受到的消息" + new String(message.getBody()));//进行手动应答/** 1.消息的标记 tag，表明消息的唯一标识* 2.是否批量应答 应该不允许批量，防止消息的丢失* */channel.basicAck(message.getEnvelope().getDeliveryTag(), false);};//消息被取消时执行下面的内容CancelCallback cancelCallback = consumerTag -> {System.out.println(consumerTag + "消息消费被中断");};//消费者进行消息的消费channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, cancelCallback); //采用手动应答（false）}
}

4.效果展示

正常情况下消息发送方发送两个消息消费者1 和消费者2 分别接收到消息并进行处理

消息自动重新入队

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

效果演示：

将上述代码中消费者2(睡眠时间更长)在消费消息的途中停止

发送者发送消息，发出消息之后的把消费者2 停掉，按理说该消费者2 来处理该消息，但是由于它处理时间较长，在还未处理完，也就是说消费者2 还没有执行 ack 代码的时候，消费者2 被停掉了，此时会看到消息被消费者1 接收到了，说明消息被重新入队，然后分配给能处理消息的消费者1处理了

【RabbitMQ】消息应答--ack机制相关推荐

RabbitMQ消息应答------ack机制
Message acknowledgment(消息应答) 执行一个任务可能需要花费几秒钟,你可能会担心如果一个消费者在执行任务过程中挂掉了.一旦RabbitMQ将消息分发给了消费者,就会从内存中删除. ...
rabbitmq消息队列 ack机制(消息确认机制)和消息补偿机制
参考:https://blog.csdn.net/pan_junbiao/article/details/112956537 ack 机制就是消息在生产者在发布消息以后,消息存在内存中,如果消息被确 ...
4-5：TCP协议之确认应答(ACK)机制和超时重传机制
文章目录一:TCP的确认应答(ACK)机制二:超时重传机制一:TCP的确认应答(ACK)机制在TCP中,当发送端的数据达到接收主机时,接收端主机会返回一个已收到消息的通知,这个消息叫做ACK( ...
RabbitMQ的消息确认ACK机制
1.什么是消息确认ACK. 答:如果在处理消息的过程中,消费者的服务器在处理消息的时候出现异常,那么可能这条正在处理的消息就没有完成消息消费,数据就会丢失.为了确保数据不会丢失,RabbitMQ支持消 ...
RabbitMQ：消费者ACK机制、生产者消息确认
文章目录基础案例环境搭建: 环境: 1. 生产者发送消息确认 1.1 confirm 确认模式 1.2 return 退回模式源代码 1.1.3 小结 2. 消费者签收消息(ACK) 2.1 代码 ...
[RabbitMQ]消息应答概念_消息手动应答代码
消息应答概念消费者完成一个任务可能需要一段时间,如果其中一个消费者处理一个长的任务并仅只完成了部分突然它挂掉了,会发生什么情况.RabbitMQ 一旦向消费者传递了一条消息,便立即将该消息标记为 ...
RabbitMQ消息应答
消费者完成一个任务可能需要一段时间,如果其中一个消费者处理一个长的任务并仅只完成了部分突然它挂掉了,会发生什么情况.RabbitMQ一旦向消费者传递了一条消息,便立即将该消息标记为删除.在这种情况下, ...
RabbitMQ消息应答实战（针对自动|手动应答常见问题进行模拟）
消息应答概念消费者完成一个任务可能需要一段时间,如果其中一个消费者处理一个长的任务并仅只完成了部分突然它挂掉了,会发生什么情况.RabbitMQ 一旦向消费者传递了一条消息,便立即将该消息标记为删除 ...
RabbitMQ中的消息确认ACK机制
我们将消息持久化后,假如消费端出现异常,rabbitmq服务器会将消息缓存到内存,当生产者发送一直发送消息而消费者都没有正常消费时消息就会将这些消息全部保存在内存,当我们的消息过多时,就可能导致rab ...

【RabbitMQ】消息应答--ack机制

消息应答

概念

自动应答

消息应答的方法

Multiple 的解释

手动应答实现

1.准备工具类

2.生产者

3.两个睡眠时间不同的消费者

4.效果展示

消息自动重新入队

效果演示：

【RabbitMQ】消息应答--ack机制相关推荐

最新文章

热门文章