网站制作工作室,网业截屏怎么截,阿里云网站域名备案,营销网站的方法消息的应答机制
消费者完成一个任务可能需要一段时间#xff0c;如果其中一个消费者处理一个长的任务并仅只完成了部分突然它挂掉了#xff0c;会发生什么情况。RabbitMQ 一旦向消费者传递了一条消息#xff0c;便立即将该消息标记为删除。在这种情况下#xff0c;突然有个…消息的应答机制
消费者完成一个任务可能需要一段时间如果其中一个消费者处理一个长的任务并仅只完成了部分突然它挂掉了会发生什么情况。RabbitMQ 一旦向消费者传递了一条消息便立即将该消息标记为删除。在这种情况下突然有个消费者挂掉了我们将丢失正在处理的消息。以及后续发送给该消费者的消息因为它无法接收到。
为了保证消息在发送过程中不丢失RabbitMQ 引入消息应答机制消息应答就是消费者在接收到消息并且处理该消息之后告诉 RabbitMQ 它已经处理了 RabbitMQ 可以把该消息删除了。
自动应答
消息发送后立即被认为已经传送成功这种模式需要在高吞吐量和数据传输安全性方面做权衡因为这种模式如果消息在接收到之前消费者那边出现连接或者 channel 关闭那么消息就丢失了当然另一方面这种模式消费者那边可以传递过载的消息没有对传递的消息数量进行限制当然这样有可能使得消费者这边由于接收太多还来不及处理的消息导致这些消息的积压最终使得内存耗尽最终这些消费者线程被操作系统杀死所以这种模式仅适用在消费者可以高效并以某种速率能够处理这些消息的情况下使用。
手动应答
手动应答的好处是可以批量应答并且减少网络拥堵批量应答的批量范围是channel 上未应答的消息。比如说 channel 上有传送 tag 的消息5678 当前 tag 是8 那么此时5-8 的这些还未应答的消息都会被确认收到消息应答。但是实际上还是不建议开启批量应答的。 Channel.basicAck(用于肯定确认)
RabbitMQ 已知道该消息并且成功的处理消息可以将其丢弃了。
Channel.basicNack(用于否定确认)
不处理该消息了直接拒绝可以将其丢弃了。
Channel.basicReject(用于否定确认)
与 Channel.basicNack 相比少一个参数。
不处理该消息了直接拒绝可以将其丢弃了。
消息自动重新入队
如果消费者由于某些原因失去连接(其通道已关闭连接已关闭或 TCP 连接丢失)导致消息未发送 ACK 确认RabbitMQ 将了解到消息未完全处理并将对其重新排队。如果此时其他消费者可以处理它将很快将其重新分发给另一个消费者。这样即使某个消费者偶尔死亡也可以确保不会丢失任何消息。
代码实现
生产者
public class MyProducer {Testpublic void test() throws Exception {// 队列名称String queue xw_queue;String message Hello World - ;// 创建工厂ConnectionFactory factory new ConnectionFactory();factory.setVirtualHost(/);factory.setHost(xuewei.world);factory.setUsername(xuewei);factory.setPassword(123456);factory.setPort(5672);// 创建连接和通道Connection connection factory.newConnection();Channel channel connection.createChannel();for (int i 0; i 20; i) {// 发布消息channel.basicPublish(xw_exchange, queue, null, (message i).getBytes());}}
}消费者1
开启手动确认后消费者1如果在处理消息的回调中不确认消息那么队列中的消息会处于unacked的状态如果消费者1突然挂掉那么这些未确认的消费会重新发送给其他消费者。
public class MyConsumer1 {public static void main(String[] args) throws Exception {// 队列名称String queue xw_queue;// 创建工厂ConnectionFactory factory new ConnectionFactory();factory.setVirtualHost(/);factory.setHost(xuewei.world);factory.setUsername(xuewei);factory.setPassword(123456);factory.setPort(5672);// 创建连接和通道Connection connection factory.newConnection();Channel channel connection.createChannel();channel.queueDeclare(queue, true, false, false, null);channel.queueBind(, xw_exchange, queue);// 配置开启手动应答channel.basicConsume(queue, false, new DefaultConsumer(channel) {SneakyThrowsOverridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {// 消费完成后手动应答// channel.basicAck(envelope.getDeliveryTag(), false);Thread.sleep(5000);System.out.println(消费者1接收到消息: new String(body));}});}
}消费者2
消费者2正常消费消息收到消息后立刻确认。
public class MyConsumer2 {public static void main(String[] args) throws Exception {// 队列名称String queue xw_queue;// 创建工厂ConnectionFactory factory new ConnectionFactory();factory.setVirtualHost(/);factory.setHost(xuewei.world);factory.setUsername(xuewei);factory.setPassword(123456);factory.setPort(5672);// 创建连接和通道Connection connection factory.newConnection();Channel channel connection.createChannel();channel.queueDeclare(queue, true, false, false, null);channel.queueBind(, xw_exchange, queue);// 配置开启手动应答channel.basicConsume(queue, false, new DefaultConsumer(channel) {SneakyThrowsOverridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {// 消费完成后手动应答channel.basicAck(envelope.getDeliveryTag(), false);System.out.println(消费者2接收到消息: new String(body));}});}
}效果展示 此时把消费者1停掉那么上面这10条为确认的消费会重新入队发送给另外的消费者。
