ack机制

每个Consumer可能需要一段时间才能处理完收到的数据。如果在这个过程中，Consumer出错了，异常退出了，而数据还没有处理完成，那么非常不幸，这段数据就丢失了。

因为我们采用no-ack的方式进行确认，也就是说，每次Consumer接到数据后，而不管是否处理完 成，RabbitMQ Server会立即把这个Message标记为完成，然后从queue中删除了。
如果一个Consumer异常退出了，它处理的数据能够被另外的Consumer处理，这样数据在这种情况下就不会丢失了（注意是这种情况下）。

为了保证数据不被丢失，RabbitMQ支持消息确认机制，即acknowledgments。为了保证数据能被正确处理而不仅仅是被Consumer收到，那么我们不能采用no-ack。而应该是在处理完数据后发送ack。即手动ACK。

在处理数据后发送的ack，就是告诉RabbitMQ数据已经被接收，处理完成，RabbitMQ可以去安全的删除它了。
如果Consumer退出了但是没有发送ack，那么RabbitMQ就会把这个Message发送到下一个Consumer。这样就保证了在Consumer异常退出的情况下数据也不会丢失。
这里并没有用到超时机制。RabbitMQ仅仅通过Consumer的连接中断来确认该Message并没有被正确处理。也就是说，RabbitMQ给了Consumer足够长的时间来做数据处理。
这样即使你通过Ctr-C中断了Recieve.cs，那么Message也不会丢失了，它会被分发到下一个Consumer。如果忘记了ack，那么后果很严重。当Consumer退出时，Message会重新分发。然后RabbitMQ会占用越来越多的内存，由于 RabbitMQ会长时间运行，因此这个“内存泄漏”是致命的。
你的 consumer 代码必须能够处理各种异常，确保只要收到一条消息，最终一定能够执行一条 ACK / NACK

去调试这种错误，可以通过以下命令打印un-acked Messages.
如果连接没有断开应用要通知服务器让消息重新发送：
可以通过channel.nack(message)来让不通过的消息再次进入消息队列。
if(body==’Hello World3!’){chnl.nack(msg);
//这样就可以让这个消息再次进入队列而不用重启服务。
}else{
console.log(‘ack’);
chnl.ack(msg);
}

---

消费端的手工ACK与NACK

当我们设置 autoACK=false 时，就可以使用手工ACK方式了，那么其实手工方式包括了手工ACK与NACK。
当我们手工 ACK 时，会发送给Broker一个应答，代表消息成功处理了，Broker就可以回送响应给生产端了。NACK 则表示消息处理失败了，如果设置重回队列，Broker端就会将没有成功处理的消息重新发送。

使用方式

消费端进行消费的时候，如果由于业务异常我们可以手工 NACK 并进行日志的记录，然后进行补偿！
方法：void basicNack(long deliveryTag, boolean multiple, boolean requeue)
requeue为true，表示deliveryTag=n之前未确认的消息都处理失败且将这些消息重新放回队列中。
requeue为false,表示deliveryTag=n之前未确认的消息都处理失败且将这些消息直接丢弃。

如果由于服务器宕机等严重问题，那我们就需要手工进行 ACK 保障消费端消费成功！
方法：void basicAck(long deliveryTag, boolean multiple)

消费端的重回队列

• 消费端重回队列是为了对没有处理成功的消息，把消息重新会递给Broker！
• 重回队列，会把消费失败的消息重新添加到队列的尾端，供消费者继续消费。
• 一般我们在实际应用中，都会关闭重回队列，也就是设置为false

演示重回队列

生产端

对消息设置自定义属性以便进行区分

public class Producer {
    public static void main(String[] args) throws Exception {
        //1 创建ConnectionFactorys
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.11.76");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        //2 获取Connection
        Connection connection = connectionFactory.newConnection();
        //3 通过Connection创建一个新的Channel
        Channel channel = connection.createChannel();
        
        String exchange = "test_ack_exchange";
        String routingKey = "ack.save";
        
        for(int i =0; i<5; i ++){
            //设置消息属性
            Map<String, Object> headers = new HashMap<String, Object>();
            headers.put("num", i);
            AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                    .deliveryMode(2)
                    .contentEncoding("UTF-8")
                    .headers(headers)
                    .build();
            //发送消息
            String msg = "Hello RabbitMQ ACK Message " + i;
            channel.basicPublish(exchange, routingKey, true, properties, msg.getBytes());
        }   
    }
}

自定义消费者

对第一条消息进行NACK，并设置重回队列

public class MyConsumer extends DefaultConsumer {

    private Channel channel ;

    public MyConsumer(Channel channel) {
        super(channel);
        this.channel = channel;
    }

    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        System.err.println("-----------consume message----------");
        System.err.println("body: " + new String(body));
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        if((Integer)properties.getHeaders().get("num") == 0) {
            //NACK，参数三requeue：是否重回队列
            channel.basicNack(envelope.getDeliveryTag(), false, true);
        } else {
            channel.basicAck(envelope.getDeliveryTag(), false);
        }
    }
}

消费端

关闭自动签收功能

public class Consumer {
    
    public static void main(String[] args) throws Exception {
        
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.11.76");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();
        
        String exchangeName = "test_ack_exchange";
        String queueName = "test_ack_queue";
        String routingKey = "ack.#";
        //声明交换机和队列，然后进行绑定设置路由Key
        channel.exchangeDeclare(exchangeName, "topic", true, false, null);
        channel.queueDeclare(queueName, true, false, false, null);
        channel.queueBind(queueName, exchangeName, routingKey);
        
        //手工签收 必须要设置 autoAck = false
        channel.basicConsume(queueName, false, new MyConsumer(channel));
    }
}

运行说明

先启动消费端，然后启动生产端，消费端打印如下，显然第一条消息由于我们调用了NACK，并且设置了重回队列，所以会导致该条消息一直重复发送，消费端就会一直循环消费

-----------consume message---------- body: Hello RabbitMQ ACK Message 0 
-----------consume message---------- body: Hello RabbitMQ ACK Message 1 
-----------consume message---------- body: Hello RabbitMQ ACK Message 2 
-----------consume message---------- body: Hello RabbitMQ ACK Message 3 
-----------consume message---------- body: Hello RabbitMQ ACK Message 4 
-----------consume message---------- body: Hello RabbitMQ ACK Message 0 
-----------consume message---------- body: Hello RabbitMQ ACK Message 0 
-----------consume message----------

一般工作中不会设置重回队列这个属性，我们都是自己去做补偿或者投递到延迟队列里的，然后指定时间去处理即可。