rabbitMQ:消费者确认模式

为了保证消息从队列可靠地到达消费者，RabbitMQ提供消息确认机制(message acknowledgment)。消费者在声明队列时，可以指定noAck参数，当noAck=false时，RabbitMQ会等待消费者显式发回ack信号后才从内存(和磁盘，如果是持久化消息的话)中移去消息。否则，RabbitMQ会在队列中消息被消费后立即删除它。

在Consumer中Confirm模式中分为手动确认和自动确认。

手动确认主要并使用以下方法：

basicAck(): 用于肯定确认，multiple参数用于多个消息确认。

basicRecover()：是路由不成功的消息可以使用recovery重新发送到队列中。

basicReject()：是接收端告诉服务器这个消息我拒绝接收,不处理,可以设置是否放回到队列中还是丢掉，而且只能一次拒绝一个消息,官网中有明确说明不能批量拒绝消息，为解决批量拒绝消息才有了basicNack。

basicNack()：可以一次拒绝N条消息，客户端可以设置basicNack方法的multiple参数为true。

消费者确认模式：自动确认

package com.it.rabbitmq.confirm;
 
import com.rabbitmq.client.*;
 
import java.io.IOException;
import java.util.concurrent.TimeoutException;
 
public class Receive {
    public static void main(String[] args)  {
        //创建链接工厂对象
        ConnectionFactory factory=new ConnectionFactory();
        factory.setUsername("root");
        factory.setPassword("123456");
        factory.setHost("192.168.174.129");
        factory.setPort(5672);
 
        Connection connection=null;//定义链接对象
        Channel channel=null;//定义通道对象
 
        try {
            connection=factory.newConnection();//实例化链接对象
            channel=connection.createChannel();//实例化通道对象
 
 
            //指定Exchange的类型
            //参数1为 交换机名称
            //参数2为交换机类型取值为 direct、queue、topic、headers
            //参数3 为是否为持久化消息 true表示持久化消息 false表示非持久化
            channel.queueDeclare("confirmQueue", true, false, false, null);
            channel.exchangeDeclare("directConfirmExchange", "direct", true);
            channel.queueBind("confirmQueue","directConfirmExchange","confirmRoutingKey");
            /**
             * 接收消息
             * 参数2为消息的确认机制，true表示自动消息确认，确认以后消息会从队列中被移除
             * 参数为false表示手动确认，必须显示的调用一些方法
             * 注意:
             * 1.如果我们只是接收了消息但是还没有来得及处理，当前应用或数据库方面出现问题
             * 但当前消息已经自动确认并删除，这就会丢失消息
             */
            channel.basicConsume("confirmQueue",true, "",new DefaultConsumer(channel) {
 
                public void handleDelivery(String consumerTag,
                                           Envelope envelope,
                                           AMQP.BasicProperties properties,
                                           byte[] body) throws IOException {
                    //获取消息数据
                    String bodyStr = new String(body);
                    System.out.println("消费者处理了消息："+bodyStr);
                }
            });
 
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
 
    }
}

 进入控制台查看队列，消息已经被全部取出

消费者确认模式：手动确认

好处是不会因为程序有问题，导致自动确认消息后，造成部分数据丢失。

package com.it.rabbitmq.exchange.confirm.addConfirmListener;
 
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmListener;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
 
import java.io.IOException;
import java.util.concurrent.TimeoutException;
 
public class Send {
    public static void main(String[] args) {
        //创建连接工厂
        ConnectionFactory factory=new ConnectionFactory();
        //配置rabbitMQ的连接信息
        factory.setHost("192.168.174.129");
        factory.setPort(5672);
        factory.setUsername("root");
        factory.setPassword("123456");
        //定义连接
        Connection connection=null;
        //定义通道
        Channel channel=null;
 
        try {
            connection=factory.newConnection();
            channel=connection.createChannel();
            channel.queueDeclare("confirmQueue",true,false,false,null);
            channel.exchangeDeclare("directConfirmExchange","direct",true);
            channel.queueBind("confirmQueue","directConfirmExchange","confirmRoutingKey");
            String message="普通发送者确认模式的测试消息！";
            //启动发送者确认模式
            channel.confirmSelect();
//异步消息确认监听器，需要在消息发送前启动
            channel.addConfirmListener(new ConfirmListener() {
                //消息确认以后的回调方法
                //参数1：被确认的消息编号，从1开始自动递增，用于标记当前是第几个消息
                //参数2：当前消息是否同时确定了多个
                //参数2为true,表示小于等于当前消息编号的所有消息全部被确认，为false，表示只确定了当前消息
                public void handleAck(long l, boolean b) throws IOException {
                    System.out.println("消息被确认了--消息编号"+l+"   是否确认了多条："+b);
                }
                //消息没有确认的回调方法
                //如果这个方法被执行，表示当前消息没有被确认，需要进行消息的补发
                //参数1：没有确认的消息编号，从1开始自动递增，用于标记当前是第几个消息
                //参数2：当前消息是否同时没有确定了多个
                //如果参数2位true,表示小于当前编号的所有消息都没有发送成功，需要进行消息的补发
                //为false,表示当前编号消息没有发送成功，需要进行补发
                public void handleNack(long l, boolean b) throws IOException {
                    System.out.println("消息没有被确认了--消息编号"+l+"   是否没有确认多条："+b);
                }
            });
            for (int i=0;i<100;i++){
                channel.basicPublish("directConfirmExchange","confirmRoutingKey",null,message.getBytes("utf-8"));
 
            }
 
            System.out.println("消息发送成功");
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        } finally {
            if (channel!=null){
                try {
                    channel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                } catch (TimeoutException e) {
                    e.printStackTrace();
                }
            }
            if (connection!=null){
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
 
        }
 
 
    }
}

运行代码

成功的全部取出

如果不加basicAck方法会导致消息确认会全部转入上图的unacked队列，表示消息可以重复使用