大家有没有想过一件事，那就是我们发送的消息并不是每次都能达到指定的队列，比如说可能我们一不小心就发送消息的路由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配置之后，需要结合自己的业务本身来灵活使用他们俩，我给的第二个示例中是两个都使用了，这样就能够做到有始有终，即使消息不能达到备份交换器的队列，也可以返回给消息发送者，这样消息发送者就能根据自身业务对消息进行处理。
当然了，你完全可以只使用其中一个，或者都不使用。
网友评论