Return消息机制

Return Listener 用于处理一些不可路由的消息！
我们的消息生产者，通过指定一个Exchange和RoutingKey。把消息送达到某一个队列中去，然后我们的消费者监听队列，进行消费处理操作。
但是在某些情况下，如果我们在发送消息的时候，当前的Exchange不存在或者指定的路由Key路由不到，这个时候如果我们需要监听这种不可达的消息。就要使用 Return Listener!
在基础API中有一个关键的配置项：
- Mandatory : 如果为true. 则监听器会接收到路由不可达的消息，然后进行处理。如果为false,那么broker端自动删除该消息。

image.png



/**
 * return listener producer
 *
 * @author yangHX
 * createTime  2019/3/21 22:20
 */
public class Producer {

    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory connectionFactory = RabbitMqUtil.getConnectionFactory();
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();

        String exchange = "test_return_exchange";
        String routingKey = "return.save";
        String routingKeyError = "abc.save";

        String msg = "Hello RabbitMQ Return Message";

        //return listener
        channel.addReturnListener(new ReturnListener() {
            @Override
            public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("------------handle return ---------------");
                System.err.println("replyCode   :" + replyCode);
                System.err.println("replyText   :" + replyText);
                System.err.println("exchange    :" + exchange);
                System.err.println("routingKey  :" + routingKey);
                System.err.println("properties  :" + properties);
                System.err.println("body        :" + new String(body));
            }
        });

///        channel.basicPublish(exchange, routingKey, true, null, msg.getBytes());
        ///如果消息没被接收。会被删除  mandatory
        channel.basicPublish(exchange, routingKeyError, true, null, msg.getBytes());


    }
}



/**
 * return listener  consumer
 *
 * @author yangHX
 * createTime  2019/3/21 22:19
 */
public class Consumer {

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {

        ConnectionFactory connectionFactory = RabbitMqUtil.getConnectionFactory();
        Connection connection = connectionFactory.newConnection();
        Channel channel = connection.createChannel();

        String exchangeName = "test_return_exchange";
        String routingKey = "return.#";
        String queueName = "test_return_queue";
        channel.exchangeDeclare(exchangeName, "topic", true, false, null);
        channel.queueDeclare(queueName, true, false, false, null);
        channel.queueBind(queueName, exchangeName, routingKey);

        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);

        channel.basicConsume(queueName, true, queueingConsumer);

        while (true) {
            QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
            String msg = new String(delivery.getBody());
            System.err.println("return  msg: "+msg);

        }
    }
}