医院门户网站建设方案,装修公司网站建设的意义,无锡定制化网站建设,陕西省建设厅网站劳保统筹基金消费者完成一个任务可能需要一段时间#xff0c;如果其中一个消费者处理一个长的任务并仅只完成 了部分突然它挂掉了#xff0c;会发生什么情况。RabbitMQ 一旦向消费者传递了一条消息#xff0c;便立即将该消 息标记为删除。在这种情况下#xff0c;突然有个消费者挂掉了… 消费者完成一个任务可能需要一段时间如果其中一个消费者处理一个长的任务并仅只完成 了部分突然它挂掉了会发生什么情况。RabbitMQ 一旦向消费者传递了一条消息便立即将该消 息标记为删除。在这种情况下突然有个消费者挂掉了我们将丢失正在处理的消息。以及后续 发送给该消费这的消息因为它无法接收到。 为了保证消息在发送过程中不丢失rabbitmq 引入消息应答机制消息应答就是: 消费者在接 收到消息并且处理该消息之后告诉 rabbitmq 它已经处理了rabbitmq 可以把该消息删除了。 消息应答的方法
Channel.basicAck(用于肯定确认) RabbitMQ 已知道该消息并且成功的处理消息可以将其丢弃了 Channel.basicNack(用于否定确认)
Channel.basicReject(用于否定确认) 与 Channel.basicNack 相比少一个参数 不处理该消息了直接拒绝可以将其丢弃了 自动应答 消息发送后立即被认为已经传送成功这种模式需要在 高吞吐量和数据传输安全性方面做权 衡 ,因为这种模式如果消息在接收到之前消费者那边出现连接或者 channel 关闭那么消息就丢 失了,当然另一方面这种模式消费者那边可以传递过载的消息 没有对传递的消息数量进行限制 当然这样有可能使得消费者这边由于接收太多还来不及处理的消息导致这些消息的积压最终 使得内存耗尽最终这些消费者线程被操作系统杀死 所以这种模式仅适用在消费者可以高效并 以某种速率能够处理这些消息的情况下使用。默认消息采用的是自动应答. 手动应答
手动应答的好处是可以批量应答并且减少网络拥堵multiple 的 true 和 false 代表不同意思 true 代表批量应答 channel 上未应答的消息 比如说 channel 上有传送 tag 的消息 5,6,7,8 当前 tag 是 8 那么此时 5-8 的这些还未应答的消息都会被确认收到消息应答 false 同上面相比 只会应答 tag8 的消息 5,6,7 这三个消息依然不会被确认收到消息应答 消费者1
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;public class Worker01 {private static final String QUEUE_NAMEhello;public static void main(String[] args) throws Exception {Channel channel RabbitMqUtils.getChannel();System.out.println(C1 等待接收消息处理时间较短);DeliverCallback deliverCallback(consumerTag, delivery)-{String receivedMessage new String(delivery.getBody());System.out.println(接收到消息:receivedMessage);try {// 睡眠1秒Thread.sleep(1000*1);} catch (InterruptedException e) {throw new RuntimeException(e);}// 消息标记tag2.false代表只应答接收到的那个传递的消息true为应答所有消息包括传递过来的消息channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);};CancelCallback cancelCallback(consumerTag)-{System.out.println(consumerTag消费者取消消费接口回调逻辑);};System.out.println(C1 消费者启动等待消费......);boolean autoAckfalse;channel.basicConsume(QUEUE_NAME,autoAck,deliverCallback,cancelCallback);}
} 消费者2 import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;public class Worker02 {private static final String QUEUE_NAMEhello;public static void main(String[] args) throws Exception {Channel channel RabbitMqUtils.getChannel();System.out.println(C2 等待接收消息处理时间较长);DeliverCallback deliverCallback(consumerTag, delivery)-{String receivedMessage new String(delivery.getBody());System.out.println(接收到消息:receivedMessage);try {// 睡眠30秒Thread.sleep(1000*30);} catch (InterruptedException e) {throw new RuntimeException(e);}// 消息标记tag2.false代表只应答接收到的那个传递的消息true为应答所有消息包括传递过来的消息channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);};CancelCallback cancelCallback(consumerTag)-{System.out.println(consumerTag消费者取消消费接口回调逻辑);};System.out.println(C2 消费者启动等待消费......);boolean autoAckfalse;channel.basicConsume(QUEUE_NAME,autoAck,deliverCallback,cancelCallback);}
}
生产者
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class Producer {private final static String QUEUE_NAME hello;public static void main(String[] args) throws Exception {//创建一个连接工厂ConnectionFactory factory new ConnectionFactory();factory.setHost(43.139.59.28);factory.setUsername(guest);factory.setPassword(guest);//channel 实现了自动 close 接口 自动关闭 不需要显示关闭try(Connection connection factory.newConnection(); Channel channel connection.createChannel()) {/*** 生成一个队列* 1.队列名称* 2.队列里面的消息是否持久化 默认消息存储在内存中* 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费* 4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true 自动删除* 5.其他参数*/channel.queueDeclare(QUEUE_NAME,false,false,false,null);String messagehello world!!!!;/*** 发送一个消息* 1.发送到那个交换机* 2.路由的 key 是哪个* 3.其他的参数信息* 4.发送消息的消息体*/for (int i 0; i 10; i) {channel.basicPublish(,QUEUE_NAME,null,message.getBytes());}System.out.println(消息发送完毕);}}
}
结果 在发送者发送消息 发出消息之后的把 C2 消费者停掉按理说该 C2 来处理该消息但是 由于它处理时间较长在还未处理完也就是说 C2 还没有执行 ack 代码的时候C2 被停掉了 此时会看到消息被 C1 接收到了说明消息 被重新入队然后分配给能处理消息的 C1 处理了 消息自动重新入队 如果消费者由于某些原因失去连接(其通道已关闭连接已关闭或 TCP 连接丢失)导致消息 未发送 ACK 确认RabbitMQ 将了解到消息未完全处理并将对其重新排队。如果此时其他消费者 可以处理它将很快将其重新分发给另一个消费者。这样即使某个消费者偶尔死亡也可以确 保不会丢失任何消息。
