【上海校区】rabbitmq之ACK消息确认机制 - 黑马程序员技术交流社区

消费者确认，又可以叫消费者应答，它指的是RabbitMQ需要确认消息到底有没有被收到

boolean autoAck = true;
channel.basicConsume(QUEUE, autoAck, defaultConsumer);

在订阅消息的时候可以指定应答模式,当自动应答等于true的时候，表示当消费者一收到消息，消息就是已经被消费了，消费者收到了消息就会立即从队列中删除。

[Java] 纯文本查看 复制代码

public class Producer1 {

    //队列名称

    private static final String QUEUE="helloWord";

    public static void main(String[] args) throws IOException,TimeoutException {

        Connection connection=null;

        Channel channel=null;



        try {

            //创建连接工程

            ConnectionFactory connectionFactory = new ConnectionFactory();

            //rabbitmq服务所在ip

            connectionFactory.setHost("192.168.56.130");

            //rabbitmq的连接端口

            connectionFactory.setPort(5672); 

            connectionFactory.setUsername("guest");//账户名

            connectionFactory.setPassword("guest");//密码

            //当前账户绑定的虚拟主机

            connectionFactory.setVirtualHost("/");

            //获取连接

            connection = connectionFactory.newConnection();

            //创建与交换机的通道,可以创建多个,每一个通道代表一个会话任务

            channel  = connection.createChannel();



   //绑定队列

            channel.queueDeclare(QUEUE,true,false,false,null);

            String message="hello: "+System.currentTimeMillis();



   //发送消息

            channel.basicPublish("",QUEUE,null,message.getBytes());

            System.out.println("send+++++++++++++++++");

        } catch (Exception e) {

            e.printStackTrace();

        } finally {

            channel.close();

            connection.close();

        }

    }

}

[Java] 纯文本查看 复制代码

public class consumer1 {



    //队列名称

    private static final String QUEUE="helloWord";



    public static void main(String[] args) throws IOException,TimeoutException {

        //创建连接工程

        ConnectionFactory connectionFactory = new ConnectionFactory();

        connectionFactory.setHost("192.168.56.130");

        connectionFactory.setPort(5672);

        connectionFactory.setUsername("guest");

        connectionFactory.setPassword("guest");

        connectionFactory.setVirtualHost("/");

        Connection connection = connectionFactory.newConnection();

        final Channel channel = connection.createChannel();

        //声明队列

        channel.queueDeclare(QUEUE, true, false, false, null);



        //定义消费方法

        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {

          @Override

          public void handleDelivery(String consumerTag,Envelope envelope,

              AMQP.BasicProperties properties,byte[] body)throws IOException



{

             //交换机

             String exchange = envelope.getExchange();

             //路由key

             String routingKey = envelope.getRoutingKey();

             //消息id

             long deliveryTag = envelope.getDeliveryTag();

             //消息内容

             String msg = new String(body, "utf-8");

             System.out.println("receive message.." + msg);

            }

        };



        /**

         * 监听队列：

         *  1.队列名称

         *  2.是否自动回复，设置为true自动进行回复，mq接收到回复会删除消息

         *                  设置为false需要手动进行确认消息

         *  3. 消费消息的方法，消费者接收到消息后调用此方法，获取消息

         */

        boolean autoAck = true;

channel.basicConsume(QUEUE, autoAck, defaultConsumer);





    }

}

运行生产者：

我们运行生产者代码，运行5次，我们可以看到Ready=5, Unacked=0, Total=5, Total代表队列中的消息总条数，Ready代表消费者还可以读到的条数，Unacked:代表还有多少条没有被应答。

运行消费者：

在消费者端的获取消息的第一行打个断点，可以看到，第一次进入到handleDelivery()方法时，队列瞬间被清空。Ready=0, Unacked=0, Total=0

Rabbitmq控制台：

当消费者连接上队列了，因为没有指定消费者一次获取消息的条数，所以队列把队列中的所有消息一下子推送到消费者端，当消费者订阅的该队列，消息就会从队列推到客户端，当消息从队列被推出的时的那一刻就表示已经对消息进行自动确认了，消息就会从队列中删除。

服务器宕机消息丢失：

继续运行程序成功消费了一条消息，程序重新进入handleDelivery()方法消费第二条消息，这个时候服务器突然宕机，那么你就会丢失余下的所有消息。

模拟服务器宕机：

打开任务管理器，点击详细信息，找到java.exe的进程，结束它。

回到我们编辑器，发现程序已经终止，但是消息只消费了一条，查看rabbitmq的管理界面，发现消息已经被删除，我们丢失了4条消息，这种情况将会产生严重的生产事故。

那我们如何去解决这种问题的，这时候我们就需要使用到手动应答。

[Java] 纯文本查看 复制代码

boolean autoAck = false;

channel.basicConsume(QUEUE, autoAck, defaultConsumer);



手动应答和自动应答不一样，需要将autoAck设置为false，当消费者收到消息在合适的时候来显示的进行确认，说我已经接收到了该消息了，RabbitMQ可以从队列中删除该消息了。



可以调用channel.basicAck(envelope.getDeliveryTag(), false);告诉rabbitmq删除消息，该方法添加在消息消费后，如下代码：



public class consumer1 {

   //队列名称

   private static final String QUEUE="helloWord";

   public static void main(String[] args) throws IOException, TimeoutException {

     //创建连接工程

     ConnectionFactory connectionFactory = new ConnectionFactory();

     connectionFactory.setHost("192.168.56.130");

     connectionFactory.setPort(5672);

     connectionFactory.setUsername("guest");

     connectionFactory.setPassword("guest");

     connectionFactory.setVirtualHost("/");

     Connection connection = connectionFactory.newConnection();

     final Channel channel = connection.createChannel();

      //声明队列

     channel.queueDeclare(QUEUE, true, false, false, null);

     //定义消费方法

     DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {

        @Override

        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

               //交换机

               String exchange = envelope.getExchange();

               //路由key

               String routingKey = envelope.getRoutingKey();

               //消息id

               long deliveryTag = envelope.getDeliveryTag();

               //消息内容

               String msg = new String(body, "utf-8");

               System.out.println("receive message.." + msg);



      //消费者在消费消息后进行手动确认

               channel.basicAck(envelope.getDeliveryTag(), false);

            }

        };

        boolean autoAck = false;

        channel.basicConsume(QUEUE, autoAck, defaultConsumer);

    }

}

运行5次生产者生产5条消息，在消费者端的获取消息的第一行打个断点，代码执行完channel.basicConsume(QUEUE_NAME, false, consumer);还没有进入到handleDelivery()方法时可以看到Ready=0, Unacked=5, Total=5

当代码进入handleDelivery()方法消费了一条消息后，每执行一次channel.basicAck(envelope.getDeliveryTag(), false)，Unacked和Total就会减去1，直到两个值都为0

Rabbitmq管理控制台：

服务器宕机消息还原：

这时候如果服务器单机消息为被消费的消息将会还原，结束java.exe进程，然后查看rabbitmq控制台。服务器重启消息继续正常消费。