消费端的手工ACK与NACK
当我们设置 autoACK=false 时，就可以使用手工ACK方式了，那么其实手工方式包括了手工ACK与NACK。
当我们手工 ACK 时，会发送给Broker一个应答，代表消息成功处理了，Broker就可以回送响应给生产端了。NACK 则表示消息处理失败了，如果设置重回队列，Broker端就会将没有成功处理的消息重新发送。
使用方式
- 消费端进行消费的时候，如果由于业务异常我们可以手工
NACK并进行日志的记录，然后进行补偿！
方法：void basicNack(long deliveryTag, boolean multiple, boolean requeue) - 如果由于服务器宕机等严重问题，那我们就需要手工进行
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----------
一般工作中不会设置重回队列这个属性，我们都是自己去做补偿或者投递到延迟队列里的，然后指定时间去处理即可。
