RabbitMQ的ACK确认机制保障消费端消息的可靠性详解

 public static void main(String[] args) throws Exception{
        ConnectionFactory connectionFactory=new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        //设置虚拟主机
        connectionFactory.setVirtualHost("/");
        //创建一个链接
        Connection connection = connectionFactory.newConnection();
        //创建channel
        Channel channel = connection.createChannel();
        String exchangeName="test_ack_exchange";
        String routeKey="ack.test";
        for (int i=0;i<5;i++){
            Map<String, Object> headers = new HashMap<String, Object>();
            //演示重回队列机制，使用num==0的消息签收失败重回队列
            headers.put("num", i);
            AMQP.BasicProperties properties=new AMQP.BasicProperties().builder()
                    .deliveryMode(2)
                    .contentEncoding("UTF-8")
                    .headers(headers)
                    .build();
            String msg="RabbitMQ send message ack test!"+i;
            channel.basicPublish(exchangeName,routeKey,properties,msg.getBytes());
        }
    }

public static void main(String[] args) throws  Exception{
        System.out.println("======消息接收start==========");
        ConnectionFactory connectionFactory=new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        //设置虚拟主机
        connectionFactory.setVirtualHost("/");
        //创建链接
        Connection connection = connectionFactory.newConnection();
        //创建channel
        Channel channel = connection.createChannel();
        String exchangeName="test_ack_exchange";
        String exchangeType="topic";
        //声明Exchange
        channel.exchangeDeclare(exchangeName,exchangeType,true,false,false,null);
        String queueName="test_ack_queue";
        //声明队列
        channel.queueDeclare(queueName,true,false,false,null);
        String routeKey="ack.#";
        //绑定队列和交换机
        channel.queueBind(queueName,exchangeName,routeKey);
        /**
         * autoAck：false  设置为手工签收
         */
        channel.basicConsume(queueName, false, new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("接收到消息：："+new String(body));
                    try {
                        Thread.sleep(3000); //休眠5秒
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    //演示重回队列机制，使用num==0的消息签收失败重回队列
                    if((Integer)properties.getHeaders().get("num") == 0) {
                        /**
                         * 参数说明：1、消息标识  2、是否批量签收  3、是否重回队列
                         */
                        channel.basicNack(envelope.getDeliveryTag(), false, true);
                    } else {
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            });
    }