RabbitMQ 备份交换器使用

傷城~ 2023-08-17 17:20 24阅读 0赞

生产者在发送消息时**不设置mandatory** 参数，那么消息达到路由器后匹配不到相应的队列后消息将会丢失。  
设置了mandatory 参数，那么需要添加ReturnListener的编程逻辑。  
如果既不想复杂化生产者的编程逻辑，又不想消息丢失，那么可以使用备份交换器。

顾名思义 备份交换器就是当第一个交换器未能有效匹配到队列时，路由到备份交换器，再由备份交换器区匹配队列

**模型图**  
从模型图中可以看到消息发送到名字为TestAE的路由器中，但是因为没有跟队列匹配，这个时候消息就会被发送到名字为exchange-unroute的备份交换器，这个交换器一般会为fanout型，随后就会被路由到AE-queue队列  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70]  
备份交换器需要注意的是：

*  如果交换器的名字之前声明过了在RabbitMQ中已经存在，便不可修改添加参数，修改会报错。  
    解决方案：换个交换器的名字或者删除之前的交换器
 *  注意点：
 *  如果设置的备份交换器不存在，客户端和RabbitMQ 服务端都不会有异常出现，此时消息会丢失。
 *  如果备份交换器没有绑定任何队列，客户端和RabbitMQ 服务端都不会有异常出现，此时消息会丢失。
 *  如果备份交换器没有任何匹配的队列，客户端和RabbitMQ 服务端都不会有异常出现，此时消息会丢失。
 *  如果备份交换器和mandatory 参数一起使用，那么mandatory 参数无效。

重要代码

//存储备份交换器的参数map
    Map<String, Object> spare = new HashMap<String , Object>(2);
    spare.put("alternate-exchange" , MY_SPARE);
    //声明了一个direct 类型的交换器，并且添加存储备份交换器的map参数
    channel.exchangeDeclare(EXCHANGE_NAME,"direct",true,false,spare);

代码实现：

public class SendSpare { 
    
        private final static String QUEUE_NAME = "wsd_test";
        private final static String QUEUE_NAME_2 = "wsd_test2";
        private final static String EXCHANGE_NAME = "wsd_exchange";
        private final static String ROUTING_KEY = "wsd_exchange";
        private final static String EXCHANGE_KEY = "wsd_exchange";
        private final static String MY_SPARE = "mySpare";
        private static Connection connection =null;
        private static Channel channel = null;
        public static void main(String[] args) { 
            Map<String, Object> spare = new HashMap<String , Object>(2);
            spare.put("alternate-exchange" , MY_SPARE);
            try{ 
                // 获取到连接以及mq通道
                connection = ConnectionUtil.getConnection();
                // 从连接中创建通道
               channel = connection.createChannel();
               //声明了一个direct 类型的交换器
               channel.exchangeDeclare(EXCHANGE_NAME,"direct",true,false,spare);
               //声明一个备胎路由器
                channel.exchangeDeclare(MY_SPARE,"fanout",true,false,null);
                // 声明（创建）队列
                channel.queueDeclare(QUEUE_NAME, false, false, false, null);
                //将路由与队列绑定，再为绑定的路径赋值一个绑定键
                channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,ROUTING_KEY);
    
                // 声明（创建）队列
                channel.queueDeclare(QUEUE_NAME_2, false, false, false, null);
                //绑定备胎路由器
                channel.queueBind(QUEUE_NAME_2,MY_SPARE,"");
                //发送数据
                for (int i=0;i<10;i++){ 
                    // 消息内容
                    String message = "Hello World!"+i;
                    //指定发送消息到哪个路由，以及他的路由键,消息等
                    if (i%2==0){ 
                        channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY,null, message.getBytes());
                    }else { 
                        //匹配不到队列
                        channel.basicPublish(EXCHANGE_NAME, "kkkk",null, message.getBytes());
                    }
                    System.out.println(" [x] Sent '" + message + "'");
                    Thread.sleep(200);
                }
            }catch (Exception e){ 
                e.printStackTrace();
            }finally { 
                //关闭通道和连接
                try { 
                    channel.close();
                    connection.close();
                } catch (IOException e) { 
                    e.printStackTrace();
                } catch (TimeoutException e) { 
                    e.printStackTrace();
                }
            }
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70]: /images/20230808/2193a5be1f194160b5cb242450faab6d.png