消费端ACK和重回队列

使用场景

消费端ACK和重回队列
消费端ACK使用场景：
1.消费端进行消费的时候，若是因为业务异常咱们能够进行日志记录，而后进行补偿。
2.因为服务器宕机等严重问题，那咱们就须要手工进行ACK保障消费端消费成功。

消费端的重回队列
消费端的重回队列是为了对没有处理成功的消息，把消息从新投递给broker
通常在实际应用中，都会关闭重回队列，也就是设置为false

建立生产者

package com.dwz.rabbitmq.ack;

import java.io.IOException; import java.util.HashMap; import java.util.Map; import java.util.concurrent.TimeoutException; import com.dwz.rabbitmq.util.ConnectionUtils; import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; public class Producer { public static void main(String[] args) throws IOException, TimeoutException { Connection connection = ConnectionUtils.getConnection(); Channel channel = connection.createChannel(); String exchangeName = "test_ack_exchange"; String routingkey = "ack.save"; for(int i = 0; i < 5; i++) { Map<String, Object> headers = new HashMap<>(); headers.put("num", i); AMQP.BasicProperties properties = new AMQP.BasicProperties().builder() .deliveryMode(2) .contentEncoding("utf-8") .headers(headers) .build(); String msg = "Hello rabbitmq ack-"+i+" message!"; channel.basicPublish(exchangeName, routingkey, properties, msg.getBytes()); } channel.close(); connection.close(); } }

建立消费者

package com.dwz.rabbitmq.ack;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.dwz.rabbitmq.util.ConnectionUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.AMQP.BasicProperties;

public class Consumer {
    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();
        Channel channel = connection.createChannel();
        
        String exchangeName = "test_ack_exchange";
        String routingkey = "ack.#";
        String queueName = "test_ack_queue";
        
        channel.exchangeDeclare(exchangeName, "topic", true, false, null);
        channel.queueDeclare(queueName, true, false, false, null);
        channel.queueBind(queueName, exchangeName, routingkey);
        
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
                    throws IOException {
                super.handleDelivery(consumerTag, envelope, properties, body);
                
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                
                System.out.println("消费端：" + new String(body));
                if((Integer)properties.getHeaders().get("num") == 0) {
                    //设置重回队列
                    channel.basicNack(envelope.getDeliveryTag(), false, true);
                } else {
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };
        
        //.手工签收必须关闭autoAck设置为false
        channel.basicConsume(queueName, false, consumer);
    }
}

 相关文章

RabbitMQ 消费端限流、TTL、死信队列