重回队列模式,是当投递消息失败时,让该消息重新回到队列的模式,该模式需要手动签收,并需要在消费者中进行判断,调用重回队列的确认模式
消费者
package com.flying.rabbitmq.api.ack; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; public class Consumer { public static void mainString[] args) throws Exception { ConnectionFactory connectionFactory = new ConnectionFactory); connectionFactory.setHost"127.0.0.1"); connectionFactory.setPort5672); connectionFactory.setVirtualHost"/"); Connection connection = connectionFactory.newConnection); Channel channel = connection.createChannel); String exchangeName = "test_ack_exchange"; String queueName = "test_ack_queue"; String routingKey = "ack.#"; channel.exchangeDeclareexchangeName, "topic", true, false, null); channel.queueDeclarequeueName, true, false, false, null); channel.queueBindqueueName, exchangeName, routingKey); // 手工签收 必须要关闭 autoAck = false channel.basicConsumequeueName, false, new MyConsumerchannel)); } }
自定义消费者:
package com.flying.rabbitmq.api.ack; import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Channel; import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.Envelope; import java.io.IOException; public class MyConsumer extends DefaultConsumer { private Channel channel ; public MyConsumerChannel channel) { superchannel); this.channel = channel; } @Override public void handleDeliveryString consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.err.println"-----------consume message----------"); System.err.println"body: " + new Stringbody)); try { Thread.sleep2000); } catch InterruptedException e) { e.printStackTrace); } ifInteger)properties.getHeaders).get"num") == 0) { channel.basicNackenvelope.getDeliveryTag), false, true); } else { channel.basicAckenvelope.getDeliveryTag), false); } } }
生产者:
package com.flying.rabbitmq.api.ack; import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import java.util.HashMap; import java.util.Map; public class Producer { public static void mainString[] args) throws Exception { ConnectionFactory connectionFactory = new ConnectionFactory); connectionFactory.setHost"127.0.0.1"); connectionFactory.setPort5672); connectionFactory.setVirtualHost"/"); Connection connection = connectionFactory.newConnection); Channel channel = connection.createChannel); String exchange = "test_ack_exchange"; String routingKey = "ack.save"; forint i =0; i<5; i ++){ Map<String, Object> headers = new HashMap<String, Object>); headers.put"num", i); AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder) .deliveryMode2) .contentEncoding"UTF-8") .headersheaders) .build); String msg = "Hello RabbitMQ ACK Message " + i; channel.basicPublishexchange, routingKey, true, properties, msg.getBytes)); } } }