RabbitMQ总结(一)--消息队列RabbitMQ应答模式（自动、手动）

﹏ヽ暗。殇╰゛Y 2022-11-01 05:22 225阅读 0赞

[原文链接][Link 1]

# [消息队列RabbitMQ应答模式（自动、手动）][RabbitMQ] #

为了确保消息不会丢失，RabbitMQ支持消息应答。消费者发送一个消息应答，告诉RabbitMQ这个消息已经接收并且处理完毕了。RabbitMQ就可以删除它了。  
如果一个消费者挂掉却没有发送应答，RabbitMQ会理解为这个消息没有处理完全，然后交给另一个消费者去重新处理。这样，你就可以确认即使消费者偶尔挂掉也不会丢失任何消息了。  
没有任何消息超时限制；只有当消费者挂掉时，RabbitMQ才会重新投递。即使处理一条消息会花费很长的时间。  
消息应答是默认打开的。我们通过显示的设置autoAsk=true关闭这种机制。现即自动应答开，一旦我们完成任务，消费者会自动发送应答。通知RabbitMQ消息已被处理，可以从内存删除。如果消费者因宕机或链接失败等原因没有发送ACK（不同于ActiveMQ，在RabbitMQ里，消息没有过期的概念），则RabbitMQ会将消息重新发送给其他监听在队列的下一个消费者。

案例:  
生产者端代码不变，消费者端代码这部分就是用于开启手动应答模式的。  
channel.basicConsume(QUEUE\_NAME, false, defaultConsumer);  
注：第二个参数值为false代表关闭RabbitMQ的自动应答机制，改为手动应答。  
在处理完消息时，返回应答状态，true表示为自动应答模式。  
channel.basicAck(envelope.getDeliveryTag(), false);

应答模式： 自动签收~~~

自动应答： 不在乎消费者对消息处理是否成功，都会告诉队列删除消息。如果处理消息失败，实现自动补偿（队列投递过去 重新处理）。

手动应答： 消费者处理完业务逻辑，手动返回ack（通知）告诉队列处理完了，队列进而删除消息。

生产者代码不变，消费者：

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();
            }
    }

手动应答。此时 消息队列的消息 一直没有被清除掉

如下修改：

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();
            }
    }

这样就消费完毕了

[Link 1]: https://blog.csdn.net/art_code/article/details/90509626
[RabbitMQ]: https://www.cnblogs.com/toov5/p/9940566.html