美文网首页
RabbitMQ的ACK与重回队列机制

RabbitMQ的ACK与重回队列机制

作者: OxygenPlus | 来源:发表于2020-07-14 17:03 被阅读0次
消费端的手工ACK与NACK

当我们设置 autoACK=false 时,就开启了手工ACK模式,那么其实手工模式包括了手工ACKNACK

  • 当我们手工ACK时,会发送给Broker一个消息已成功消费的应答。

  • NACK则表示消息由于业务的异常而导致处理失败了,此时如果设置重回队列,Broker端就会将会把消费失败的消息重新添加到队列的尾端

相关方法:

  1. NACK:
    方法:void basicNack(long deliveryTag, boolean multiple, boolean requeue)
  2. 如果由于服务器宕机等严重问题,那我们就需要手工进行 ACK 保障消费端消费成功!
    方法:void basicAck(long deliveryTag, boolean multiple)
消费端的重回队列
  • 消费端重回队列是为了对没有处理成功的消息,把消息重新会递给Broker!
  • 重回队列,会把消费失败的消息重新添加到队列的尾端,供消费者继续消费。
  • 一般我们在实际应用中,都会关闭重回队列,也就是设置为false

代码演示

生产端:

public class Producer { 
    public static void main(String[] args) throws Exception {   
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.11.76");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();
        
        String exchange = "test_ack_exchange";
        String routingKey = "ack.save";
        
        for(int i =0; i<5; i ++){
            Map<String, Object> headers = new HashMap<>();
            headers.put("num", i);
            // deliveryMode = 1 不持久化,deliveryMode = 2 持久化
            AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                    .deliveryMode(2)
                    .contentEncoding("UTF-8")
                    .headers(headers)
                    .build();
            String msg = "Hello RabbitMQ ACK Message " + i;
            channel.basicPublish(exchange, routingKey, true, properties, msg.getBytes());
        }
        
    }
}

消费端:

public class Consumer {
    public static void main(String[] args) throws Exception {       
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.11.76");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();       
        
        String exchangeName = "test_ack_exchange";
        String queueName = "test_ack_queue";
        String routingKey = "ack.#";
        
        channel.exchangeDeclare(exchangeName, "topic", true, false, null);
        channel.queueDeclare(queueName, true, false, false, null);
        channel.queueBind(queueName, exchangeName, routingKey);
        
        // 手工签收 必须要关闭 autoAck = false
        channel.basicConsume(queueName, false, new MyConsumer(channel));
    }
}

MyConsumer:

public class MyConsumer extends DefaultConsumer {
    private Channel channel ;
    
    public MyConsumer(Channel channel) {
        super(channel);
        this.channel = channel;
    }

    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        System.err.println("body: " + new String(body));
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        if((Integer)properties.getHeaders().get("num") == 0) {
            channel.basicNack(envelope.getDeliveryTag(), false, true);
        } else {
            channel.basicAck(envelope.getDeliveryTag(), false);
        }       
    }
}

相关文章

网友评论

      本文标题:RabbitMQ的ACK与重回队列机制

      本文链接:https://www.haomeiwen.com/subject/fjyfhktx.html