RabbitMQ的mandatory和alternate-exchange

大家有没有想过一件事,那就是我们发送的消息并不是每次都能达到指定的队列,比如说可能我们一不小心就发送消息的路由key写错了,那这个时候我们发送的消息去哪里了呢?实际上默认是丢失了的,但这对于很重要的消息肯定是不能够接受的。怎么办呢?这时候Mandatory和alternate-exchange就可以派上用场了。

mandatory

它实际上是发送消息时可以指定的一个参数,如果我们没有指定,那么默认值就是false。
如果发送消息的时候将mandatory参数指定为true,那么当交换器无法根据自身类型和路由key找到一个符合条件的队列时,我们发送的消息就会被返回给发送者。
消息发送者可以通过调用信道对象的addReturnListener()来添加一个事件回调,用于接收找到不到队列的消息。

/**
     * 发消息时将mandatory值设置为true,保证当消息路由不到队列时,将消息进行返回,以告诉客户端
     */
    public void sendMessageWithMandator(String message) {
        try {
            //1、创建信道
            Channel channel = this.connection.createChannel();
            //2、定义交换器
            //第一个参数:交换器的名称
            //第二个参数:交换器的类型
            //第三个参数:交换器是否持久化
            //第四个参数:交换器是否自动删除
            //第五个参数:是否为内部交换器
            //第六个参数:其他一些格外参数
 channel.exchangeDeclare("msgExchange",BuiltinExchangeType.DIRECT,true,false,false,null);

            //3、定义队列
            //第一个参数:队列名称
            //第二个参数:是否持久化
            //第三个参数:是否排他
            //第四个参数:是否自动删除
            channel.queueDeclare("msgQueue",true,false,false,null);

            //3、交换器和队列进行绑定
            channel.queueBind("msgQueue","msgExchange","msgRouteKey");
           
            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(replyCode);//rabbitmq返回的状态码
                    System.out.println(replyText);//rabbitmq返回的说明
                    System.out.println(exchange);//交换器
                    System.out.println(routingKey);//路由可以
                    System.out.println(new String(body,"utf-8"));//发送的消息体
                }
            });

            //4、发送消息
            channel.basicPublish("msgExchange","backMsgRouteKey1",true,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes("utf-8"));

            //5、关闭信道资源
            //channel.close();//因为监听了事务回调,所以这里发送完消息不能立即将channel资源关闭
        }catch (ShutdownSignalException ex) {
            //连接异常关闭了,这里要进行检查,并尝试重新建立连接
            ex.printStackTrace();
        } catch (IOException ex) {
            //发生io异常需要进行处理,对应channel可能关闭了
            ex.printStackTrace();
        }
    }

alternate-exchange

alternate有“备胎”的意思,从字面上理解就是备份交换器,它用于在声明交换器时给交换器指定一个备份交换器。当交换器根据自身类型和路由key找不到一个合适的队列时,交换器就会把消息转交给备份交换器(备份交换器和普通交换器没有特殊之处,它也需要绑定队列)。如果备份交换器根据自身类型和路由key也找不到合适的队列,那么消息就可能会丢失,这里的是“可能",因为如果在发送消息的时候指定了mandatory参数为true,那么即使备份交换器找不到合适的队列,消息也会再次返回给消息发送者。
当alternate-exchange和mandatory参数同时作用时,如果消息能够正常路由到备份交换器所绑定的队列,那么消息就不会返回给消息发送者,否则就会返回。
需要注意的是,消息转发给备份交换器的路由key是与消息发送者指定的路由key是同一个。

/**
     * 发消息时将mandatory值设置为true,保证当消息路由不到队列时,将消息进行返回,以告诉客户端
     */
    public void sendMessageWithMandator(String message) {
        try {
            //1、创建信道
            Channel channel = this.connection.createChannel();
            //2、定义交换器
            //第一个参数:交换器的名称
            //第二个参数:交换器的类型
            //第三个参数:交换器是否持久化
            //第四个参数:交换器是否自动删除
            //第五个参数:是否为内部交换器
            //第六个参数:其他一些格外参数
            channel.exchangeDeclare("backExchange",BuiltinExchangeType.DIRECT,true,false,true,null);
            Map<String,Object>msgExchangeArgs=new HashMap<>();
            msgExchangeArgs.put("alternate-exchange","backExchange");
            channel.exchangeDeclare("msgExchange",BuiltinExchangeType.DIRECT,true,false,false,msgExchangeArgs);

            //3、定义队列
            //第一个参数:队列名称
            //第二个参数:是否持久化
            //第三个参数:是否排他
            //第四个参数:是否自动删除
            channel.queueDeclare("msgQueue",true,false,false,null);
            channel.queueDeclare("backQueue",true,false,false,null);

            //3、交换器和队列进行绑定
            channel.queueBind("msgQueue","msgExchange","msgRouteKey");
            channel.queueBind("backQueue","backExchange","backMsgRouteKey");

            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(replyCode);//rabbitmq返回的状态码
                    System.out.println(replyText);//rabbitmq返回的说明
                    System.out.println(exchange);//交换器
                    System.out.println(routingKey);//路由可以
                    System.out.println(new String(body,"utf-8"));//发送的消息体
                }
            });

            //4、发送消息
            channel.basicPublish("msgExchange","backMsgRouteKey1",true,MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes("utf-8"));

            //5、关闭信道资源
            //channel.close();//因为监听了事务回调,所以这里发送完消息不能立即将channel资源关闭
        }catch (ShutdownSignalException ex) {
            //连接异常关闭了,这里要进行检查,并尝试重新建立连接
            ex.printStackTrace();
        } catch (IOException ex) {
            //发生io异常需要进行处理,对应channel可能关闭了
            ex.printStackTrace();
        }
    }

说明

在掌握了mandatory参数和alternate-exchange配置之后,需要结合自己的业务本身来灵活使用他们俩,我给的第二个示例中是两个都使用了,这样就能够做到有始有终,即使消息不能达到备份交换器的队列,也可以返回给消息发送者,这样消息发送者就能根据自身业务对消息进行处理。
当然了,你完全可以只使用其中一个,或者都不使用。

©著作权归作者所有,转载或内容合作请联系作者
平台声明:文章内容(如有图片或视频亦包括在内)由作者上传并发布,文章内容仅代表作者本人观点,简书系信息发布平台,仅提供信息存储服务。