【RabbitMQ】訊息可靠性投遞（四）Queue--＞Consumer

系列文章：

• 【RabbitMQ】如何保證訊息可靠性投遞？
• 【RabbitMQ】訊息可靠性投遞（一）Producer-＞Broker
• 【RabbitMQ】訊息可靠性投遞（二）Exchange-＞Queue
• 【RabbitMQ】訊息可靠性投遞（三）Queue存盤訊息
• 【RabbitMQ】訊息可靠性投遞（四）Queue–＞Consumer

如果消費者收到訊息后沒來得及處理就發生例外，或者處理程序中發生例外，會導致④失敗，服務端應該以某種方式得知消費者對訊息的接收情況，并決定是否重新投遞這條訊息給其他消費者，

RabbitMQ提供了消費者的訊息確認機制（message acknowledgement），消費者可以自動或者手動地發送ACK給服務端沒有收到ACK的訊息，消費者斷開連接后，RabbitMQ會把這條訊息發送給其他消費者，如果沒有其他消費者，消費者重啟后會重新消費這條訊息，重復執行業務邏輯，

1.自動ACK

消費者默認采用的是自動ack（autoAck=true），所以我們可以不斷的一條一條接收訊息，

而自動ack的問題在于訊息丟失問題，當訊息到達Consumer就會給broker回傳ack，若Consumer在處理中就宕機，那么當前訊息就丟失了

注：在 Kafka 中，自動 ACK 是每隔一段時間 ACK 一次，而且訊息的清除是根據根據配置的訊息的清除策略， 所以，訊息自動 ACK 容易引起的是訊息重復消費（與 RabbitMQ 正好相反）

有沒有一種方式，等Consumer處理完訊息后，在當前訊息的ack發給服務端？手動ACK

2.手動ACK

RabbitMQ會等待消費者顯式地回復確認信號后才從佇列中移去訊息，而這種方式的問題在于，若服務器未收到消費者的ack時會一直阻塞，最終可能引起訊息大量堆積，

若是采用原生API，消費者在訂閱佇列時可以指定 autoAck=false

public class AckConsumer {
    private final static String QUEUE_NAME = "TEST_ACK_QUEUE";

    public static void main(String[] args) throws Exception {
    
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUri(ResourceUtil.getKey("rabbitmq.uri"));
        // 建立連接
        Connection conn = factory.newConnection();
        // 創建訊息通道
        final Channel channel = conn.createChannel();

        // 宣告佇列（默認交換機AMQP default，Direct）
        // String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        System.out.println(" Waiting for message....");

        // 創建消費者，并接收訊息
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                String msg = new String(body, "UTF-8");
                System.out.println("Received message : '" + msg + "'");

                if (msg.contains("拒收")){
                    // 拒絕訊息
                    // requeue：是否重新入佇列，true：是；false：直接丟棄，相當于告訴佇列可以直接洗掉掉
                    // TODO 如果只有這一個消費者，requeue 為true 的時候會造成訊息重復消費
                    channel.basicReject(envelope.getDeliveryTag(), false);
                } else if (msg.contains("例外")){
                    // 批量拒絕（拒絕deliveryTag之前的訊息）
                    // requeue：是否重新入佇列
                    // TODO 如果只有這一個消費者，requeue 為true 的時候會造成訊息重復消費
                    channel.basicNack(envelope.getDeliveryTag(), true, false);
                } else {
                    // 手工應答
                    // 如果不應答，佇列中的訊息會一直存在，重新連接的時候會重復消費
                    channel.basicAck(envelope.getDeliveryTag(), true);
                }
            }
        };

        // 開始獲取訊息，注意這里開啟了手工應答
        // String queue, boolean autoAck, Consumer callback
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }

}

在 Spring AMQP中 MessageListener 相當于消費者，可以在AcknowledgeMode列舉類中看到，關于ack的配置具體有三種選擇：

• NONE：自動ACK（默認）
• MANUAL： 手動ACK
• AUTO：如果方法未拋出例外，則發送 ack，
  • 當拋出 AmqpRejectAndDontRequeueException 例外的時候，則訊息會被拒絕，且不重新入隊，
  • 當拋出 ImmediateAcknowledgeAmqpException 例外，則消費者會發送ACK，
  • 其他的例外，則訊息會被拒絕，且 requeue = true會重新入隊，

@Bean // 構建MessageListenerContainer的Bean時配置
public SimpleMessageListenerContainer container(ConnectionFactory connectionFactory) {
    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
    
    container.setQueues(getSecondQueue(),getThirdQueue()); // 監聽的佇列 
    container.setConcurrentConsumers(1); // 最小消費者數 
    container.setMaxConcurrentConsumers(5); // 最大的消費者數量 
    container.setDefaultRequeueRejected(false); // 是否重回佇列 
    container.setAcknowledgeMode(AcknowledgeMode.AUTO); // 簽收模式！！！
    container.setExposeListenerChannel(true); 
    container.setConsumerTagStrategy(new ConsumerTagStrategy() { // 消費端的標簽策略
        public String createConsumerTag(String queue) {
            return null;
        }
    });
    return container;
}

若在SpringBoot的配置：

spring.rabbitmq.listener.direct.acknowledge-mode=manual  # 默認為NONE,自動ack
spring.rabbitmq.listener.simple.acknowledge-mode=manual

具體的呼叫代碼如下：

@Component
@RabbitListener(queues = "SECOND_QUEUE")
public class SecondConsumer {
    
    @RabbitHandler 
    // 當要手動確認時，引數中要有Channel和Message (注：Channel是rabbitmq.client.Channel不是amqp的)
    public void process(String msg,Channel channel,Message message){ 
        System.out.println("Second Queue received msg:" + msg); 
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false); // 手動ack
    }
}

3.拒絕策略

如果訊息無法處理或者消費失敗，也有兩種拒絕的方式，

• void basicReject(long deliveryTag, boolean requeue)單條拒絕
• void basicNack(long deliveryTag, boolean multiple, boolean requeue)批量拒絕

如果requeue引數設定為true，可以把這條訊息重新存入佇列，以便發給下一個消費者（當然，只有一個消費者的時候，這種方式可能會出現無限回圈重復消費的情況，可以投遞到新的佇列中，或者只列印例外日志），如果requeue為false，當訊息投遞失敗就會丟棄，

但無論消費者是發送ACK還是NACK，甚至是消費者出現例外，生產者也是完全不知情的，所以，生產者最終確定消費者有沒有消費成功的方式：

• 消費者收到訊息，處理完畢后，呼叫生產者的API
• 消費者收到訊息，處理完畢后，發送一條回應訊息給生產者