RabbitMQ mandatory参数路由器匹配不到队列返回响应

小咪咪 2023-08-17 17:19 62阅读 0赞

mandatory参数当交换器**无法根据自身的类型和路由键找到一个符合条件的队列**，那么 RabbitM 会调用Basic.Return命令将消息返回给生产者 。通过调用channel.addReturnListener 来添加ReturnListener 监听器实现。  
这里有一个坑：**保证在在发送消息之前绑定监听器，在retrun message之前channel不关闭**  
不然将无法收到返回消息  
实现效果如下  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70]  
重点代码： //绑定路由不到队列的回调方法（**一定要在发送消息之前绑定**）

channel.addReturnListener(new ReturnListener() { 
                    public void handleReturn(int replyCode, String replyText,
                                             String exchange, String routingKey,
                                             AMQP.BasicProperties properties, byte[] body) throws IOException { 
                        String message = new String(body);
                        System.out.println( "Basic.Return 返回的结果是: "+message );
                    }
                });

重点代码：发送消息是时mandatory设置为true

channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY+"Basic.Return", true,null, message.getBytes());

全部代码：

public class Send { 
        //队列名
        private final static String QUEUE_NAME = "wsd_test";
        //路由器名
        private final static String EXCHANGE_NAME = "exchange";
        //绑定键
        private final static String BINDING_KEY = "wsd_exchange";
        //路由键
        private final static String ROUTING_KEY = "wsd_exchange";
    
        private static Connection connection =null;
        private static Channel channel = null;
        public static void main(String[] args) { 
            try{ 
                // 获取到连接以及mq通道
                connection = ConnectionUtil.getConnection();
                // 从连接中创建通道
                channel = connection.createChannel();
    
                //声明了一个direct 类型的交换器
                channel.exchangeDeclare(EXCHANGE_NAME,"direct",true,false,null);
    
                // 声明队列QUEUE_NAME
                channel.queueDeclare(QUEUE_NAME, false, false, false, null);
                //将路由与队列绑定，再为绑定的路径赋值一个绑定键
                channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,BINDING_KEY);
    
                //绑定路由不到队列的回调方法
                channel.addReturnListener(new ReturnListener() { 
                    public void handleReturn(int replyCode, String replyText,
                                             String exchange, String routingKey,
                                             AMQP.BasicProperties properties, byte[] body) throws IOException { 
                        String message = new String(body);
                        System.out.println( "Basic.Return 返回的结果是: "+message );
                    }
                });
    
                //发送数据
                for (int i=0;i<10;i++){ 
                    // 消息内容
                    String message = "Hello World!"+i;
                    //指定发送消息到哪个路由，以及他的路由键,消息等
                    channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY+"Basic.Return", true,null, message.getBytes());
                    System.out.println(" [x] Sent '" + message + "'");
                    Thread.sleep(200);
                }
            }catch (Exception e){ 
                e.printStackTrace();
            }
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70]: /images/20230808/ad1f03a4eb2e42979082de424983f058.png