每个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----------
一般工作中不会设置重回队列这个属性，我们都是自己去做补偿或者投递到延迟队列里的，然后指定时间去处理即可。
网友评论