消費端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、死信隊列