消费端ACK和重回队列
消费端ACK使用场景:
1.消费端进行消费的时候,如果由于业务异常我们可以进行日志记录,然后进行补偿。
2.由于服务器宕机等严重问题,那我们就需要手工进行ACK保障消费端消费成功。
消费端的重回队列
消费端的重回队列是为了对没有处理成功的消息,把消息重新投递给broker
一般在实际应用中,都会关闭重回队列,也就是设置为false
package com.dwz.rabbitmq.ack;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;
import com.dwz.rabbitmq.util.ConnectionUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
public class Producer {
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = ConnectionUtils.getConnection();
Channel channel = connection.createChannel();
String exchangeName = "test_ack_exchange";
String routingkey = "ack.save";
for(int i = 0; i < 5; i++) {
Map<String, Object> headers = new HashMap<>();
headers.put("num", i);
AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
.deliveryMode(2)
.contentEncoding("utf-8")
.headers(headers)
.build();
String msg = "Hello rabbitmq ack-"+i+" message!";
channel.basicPublish(exchangeName, routingkey, properties, msg.getBytes());
}
channel.close();
connection.close();
}
}
package com.dwz.rabbitmq.ack; import java.io.IOException; import java.util.concurrent.TimeoutException; import com.dwz.rabbitmq.util.ConnectionUtils; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.Envelope; import com.rabbitmq.client.AMQP.BasicProperties; public class Consumer { public static void main(String[] args) throws IOException, TimeoutException { Connection connection = ConnectionUtils.getConnection(); Channel channel = connection.createChannel(); String exchangeName = "test_ack_exchange"; String routingkey = "ack.#"; String queueName = "test_ack_queue"; channel.exchangeDeclare(exchangeName, "topic", true, false, null); channel.queueDeclare(queueName, true, false, false, null); channel.queueBind(queueName, exchangeName, routingkey); DefaultConsumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { super.handleDelivery(consumerTag, envelope, properties, body); try { Thread.sleep(2000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("消费端:" + new String(body)); if((Integer)properties.getHeaders().get("num") == 0) { //设置重回队列 channel.basicNack(envelope.getDeliveryTag(), false, true); } else { channel.basicAck(envelope.getDeliveryTag(), false); } } }; //.手工签收必须关闭autoAck设置为false channel.basicConsume(queueName, false, consumer); } }
原文:https://www.cnblogs.com/zheaven/p/11833528.html