RabbitMQ消息确认以及return机制

消息确认机制

消息发送出去后，如果想确定消息发送的状态信息，发送成功或者发送失败，rabbitmq中有一套完整的检测机制来检测这类问题，我们只需要调用即可，具体的检测机制我们后面再讨论，暂且看一下怎么用代码先实现：

public class Send {private static final String EXCHANGE_NAME = "test_exchange_confirm";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = RabbitConnection.getConnection();Channel channel = connection.createChannel();channel.confirmSelect();channel.exchangeDeclare(EXCHANGE_NAME, "topic");//send a msgString msg = "hello comfirm";String routingKey = "confirm.save";channel.basicPublish(EXCHANGE_NAME, routingKey, null, msg.getBytes());channel.addConfirmListener(new ConfirmListener() {@Overridepublic void handleAck(long l, boolean b) throws IOException {System.out.println(" ack");}@Overridepublic void handleNack(long l, boolean b) throws IOException {System.out.println("no ack");}});System.out.println("send:"+msg);
//        channel.close();
//        connection.close();  这里需要注释，否则就收不到ack}
}

消息发送成功后，发送端发送成功后会收到ack，发送失败后会收到 no ack

消费端写法和之前的没什么区别，我们这次用topic的路由机制来实现一下：

public class ReceiveOne {private static final String QUEUE_NAME = "receive1_queue";private static final String EXCHANGE_NAME = "test_exchange_confirm";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = RabbitConnection.getConnection();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);channel.basicQos(1);String routingKey = "confirm.#";  //看上一节的topic路由机制channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, routingKey);DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [ConsumerOne is] Received '" + message + "'");};boolean autoAck = true;  //开启自动应答channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, consumerTag -> { });}
}

Return 机制

return Listener用于处理一些不可路由的消息（在某些情况下，如果我们发送消息的时候，当前的exchange或者routeKey路由不到的时候，这个时候如果我们需要监听这种不可到达的消息，就要使用Return Listener）

消费端自定义监听：

生产者：

public class Produces {private static final String EXCHANGE_NAME = "test_undedine_consumer";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = RabbitConnection.getConnection();Channel channel = connection.createChannel();String routingKey = "consumer.save";String msg = "Hello undefine consumer";for(int i = 0; i<5; i++){channel.basicPublish(EXCHANGE_NAME,routingKey,true,null,msg.getBytes());}}

消费者：

public class Consumers {private static final String QUEUE_NAME = "test_consumer_queue";private static final String EXCHANGE_NAME = "test_undedine_consumer";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = RabbitConnection.getConnection();Channel channel = connection.createChannel();String routineKey = "consumer.#";channel.exchangeDeclare(EXCHANGE_NAME,"topic",true,false,null);channel.queueDeclare(QUEUE_NAME,true,false,false,null);channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,routineKey);channel.basicConsume(QUEUE_NAME,true,new Myconsumer(channel));}
}

自定义内容：

public class Myconsumer extends DefaultConsumer {public Myconsumer(Channel channel) {super(channel);}@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("---------");System.out.println(consumerTag);}
}

消费端限流

生产者：（和上面的没什么区别）

public class Produces {private static final String EXCHANGE_NAME = "test_limit_flu";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = RabbitConnection.getConnection();Channel channel = connection.createChannel();String routingKey = "qos.save";String msg = "limit fluency";for(int i = 0; i<5; i++){channel.basicPublish(EXCHANGE_NAME,routingKey,true,null,msg.getBytes());}}
}

消费者：

public class Consumers {private static final String QUEUE_NAME = "test_qos";private static final String EXCHANGE_NAME = "test_limit_flu";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = RabbitConnection.getConnection();Channel channel = connection.createChannel();String routineKey = "qos.#";channel.exchangeDeclare(EXCHANGE_NAME,"topic",true,false,null);channel.queueDeclare(QUEUE_NAME,true,false,false,null);channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,routineKey);boolean autoAck = false;channel.basicQos(0,1,false);channel.basicConsume(QUEUE_NAME,autoAck,new Myconsumer(channel));}
}

区别在自定义处理里面：

public class Myconsumer extends DefaultConsumer {private Channel channel;public Myconsumer(Channel channel) {super(channel);this.channel = channel;}@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("---------");System.out.println(consumerTag);//channel.basicAck(envelope.getDeliveryTag(),false);//这里之所以是false，前面我们设置的是  channel.basicQos(0,1,false); 表示只接受一个，如果改为1以上，就要相应的改为true}
}

当我们把channel.basicAck(envelope.getDeliveryTag(),false); 这一段屏蔽时，我们在后台看到的是：

总共是五条消息，只有一条能被消费，其他的四条需要等这条消费完后发ack响应才能，继续向前传递。屏蔽的那段代码就是手动响应ACK。

重回队列

死信队列

当一个消息在队列中没人去消费它，它就会被重新publish到另外一个Exchange，这个Exchange就是死信队列。

死信队列有哪几种情况？

当消息被拒绝（basic.nack)，并且拒绝重回队列
TTL过期了
队列达到最大的长度

下面用代码实现一下：

生产端：

public class Produces {private static final String EXCHANGE_NAME = "test_dlx";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = RabbitConnection.getConnection();Channel channel = connection.createChannel();String routingKey = "dlx.save";String msg = "hello dlx";channel.basicPublish(EXCHANGE_NAME,routingKey,true,null,msg.getBytes());}
}

消费端：

public class Consumers {private static final String QUEUE_NAME = "test_dlx.queue";private static final String EXCHANGE_NAME = "test_dlx";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = RabbitConnection.getConnection();Channel channel = connection.createChannel();String routineKey = "dlx.#";channel.exchangeDeclare(EXCHANGE_NAME,"topic",true,false,null);// 下面是配置死信队列Map<String,Object> arg = new HashMap<>();arg.put("x-dead-letter-exchange","dlx.exchange");channel.queueDeclare(QUEUE_NAME,true,false,false,arg); //这里表明了路由失败的话，转发到死信队列上channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,routineKey);// 死信队列的声明channel.exchangeDeclare("dlx.exchange","topic",true,false,null);channel.queueDeclare("dlx.queue",true,false,false,null);channel.queueBind("dlx.queue","dlx.exchange","#");boolean autoAck = true;channel.basicQos(0,1,false);channel.basicConsume(QUEUE_NAME,autoAck,new Myconsumer(channel));}
}

这里需要说明的是我们定义了一个死信队列 “dlx.queue”，和dlx.exchange绑定在一起，也就是说消息没人消费就会路由到死信队列中去。

public class Myconsumer extends DefaultConsumer {private Channel channel;public Myconsumer(Channel channel) {super(channel);this.channel = channel;}@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {// channel.basicAck(envelope.getDeliveryTag(),false);//这里之所以是false，前面我们设置的是  channel.basicQos(0,1,false); 表示只接受一个，如果改为1以上，就要相应的改为trueSystem.out.println(new String(body));}
}

可以看到上面的test_dlx.queue后面有个DLX的标记，也就是说，它里面的无人消费的消息都会直接转到dlx.queue中。我们可以在dlx.queue中设置监听来处理这些消息。