RabbitMQ整合Spring Booot【消費者應答模式】

生產者代碼不變，消費者：

package com.toov5.Consumer;

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

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;
import com.toov5.utils.MQConnectionUtils;

public class Consumer {
  
     //隊列名稱
        private static final String QUEUE_NAME = "test_queue";
        
        public static void main(String[] args) throws IOException, TimeoutException {
            System.out.println("消費者啓動..........");
            //建立新的鏈接
        Connection connection = MQConnectionUtils.newConnection();
           //建立Channel
            Channel channel = connection.createChannel();
            // 消費者關聯隊列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            
              DefaultConsumer defaultConsumerr = new DefaultConsumer(channel) {
                  //監聽獲取消息
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                            byte[] body) throws IOException {
                        String msg =new String(body,"UTF-8");
                        System.out.println("消費者獲取生產者消息："+msg);
                    }
              };
            //牽手模式設置  默認自動應答模式  true:自動應答模式  
              channel.basicConsume(QUEUE_NAME, false, defaultConsumerr);//    fanse手動應答          
              
//            //關閉通道和鏈接
//             channel.close();
//             connection.close();
        }
}

手動應答。此時 消息隊列的消息 一直沒有被清除掉

生產者作以下修改就OK了：

 

package com.toov5.Consumer;

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

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;
import com.toov5.utils.MQConnectionUtils;

public class Consumer {
  
     //隊列名稱
        private static final String QUEUE_NAME = "test_queue";
        
        public static void main(String[] args) throws IOException, TimeoutException {
            System.out.println("消費者啓動..........");
            //建立新的鏈接
        Connection connection = MQConnectionUtils.newConnection();
           //建立Channel
            final Channel channel = connection.createChannel();
            // 消費者關聯隊列
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            
              DefaultConsumer defaultConsumerr = new DefaultConsumer(channel) {
                  //監聽獲取消息
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                            byte[] body) throws IOException {
                        String msg =new String(body,"UTF-8");
                        System.out.println("消費者獲取生產者消息："+msg);
                        channel.basicAck(envelope.getDeliveryTag(), false);  //手動應答 告訴消息隊列服務器 消費成功
                    }
              };
            //牽手模式設置  默認自動應答模式  true:自動應答模式  
              channel.basicConsume(QUEUE_NAME, false, defaultConsumerr);//    fanse手動應答          
              
//            //關閉通道和鏈接
//             channel.close();
//             connection.close();
        }
}