RabbitMQ中的消息不可达returnlistener和mandatory的使用

Posted 2022-02-25 lflying

return listener 用于处理一些不可路由的消息。
    我们的消息生产者，通过指定一个exchange和routingkey，把消息送达到某一个队列中，然后我们的消费者监听队列，进行消费处理操作。
    但是在某种情况下，如果我们在发送消息的时候，当前的exchange不存在或者制定的路由key路由不到，这个时候如果我们需要监听这种不可达的消息，就要使用return listener。
    mandatory， 设置为true，则监听器会接收到路由不可达的消息， 然后进行处理，如果设置为false，那么broker端自动删除该消息。

消费者：

package com.flying.rabbitmq.api.returnlistener;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.QueueingConsumer.Delivery;

public class Consumer 

    
    public static void main(String[] args) throws Exception 
        
        
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        
        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)
            
            Delivery delivery = queueingConsumer.nextDelivery();
            String msg = new String(delivery.getBody());
            System.err.println("消费者: " + msg);
        
        
        
        
        
        
    

生产者：

package com.flying.rabbitmq.api.returnlistener;

import com.rabbitmq.client.*;

import java.io.IOException;

/**
 *  当mandatory标志位设置为true时，
 *  如果exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息，
 *  那么broker会调用basic.return方法将消息返还给生产者;当mandatory设置为false时，
 *  出现上述情况broker会直接将消息丢弃;通俗的讲，mandatory标志告诉broker代理服务器至少将消息route到一个队列中，
 *  否则就将消息return给发送者;
 */
public class Producer 

    
    public static void main(String[] args) throws Exception 
        
        
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        
        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";
        
        channel.addReturnListener(new ReturnListener() 
            @Override
            public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException 

                System.err.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());