RabbitMQ处理未被路由的消息

     我们经常使用消息队列进行系统之间的解耦，日志记录等等。但是有时候我们在使用 RabbitMQ时，由于exchange、bindKey、routingKey没有设置正确，导致我们发送给交换器(exchange)的消息，由于没有正确的RoutingKey可能会存在一个消息丢失的情况，如果我们希望知道那些消息经过exchange之后，没有被正确的存入消息队列，那么应该如何进行处理。

方案一：使用 mandatory 参数配合 ReturnListener 来进行解决

方案二：使用备份交换器 (alternate exchange) 来进行解决

 

方案一介绍：

mandatory参数的含义：

 

       true：表示当交换器无法根据自身的类型和路由键找到一个符合条件的队列时，那么RabbitMQ会调用  Basic.Return 命令将消息返回给生产者。生产者使用ReturnListener 来监听没有被正确路由到消息队列中的消息。

 

       false：表示当交换器无法根据自身的类型和路由键找到一个服务条件的队列时，那么RabbitMQ会丢弃这个消息。

注意事项：

     1、有时候发现即使 mandatory参数设置成 true，也没有进入 ReturnListener，那么这个可能是什么原因呢？其实这个可能是受RabbitMQ配置的内存和磁盘告警限制。（http://www.rabbitmq.com/alarms.html）

     2、这是一个RabbitMQ配置的磁盘告警导致没有进入ReturnListener的例子。（http://rabbitmq.1065348.n5.nabble.com/ReturnListener-is-not-invoked-td24549.html）

示例代码：

/**
 * RabbitMQ 生产者
 * <pre>
 * 	   1、ReturnListener 的使用。
 * 	   		>> mandatory: 参数需要设置成 true , ReturnListener 才会生效。
 * 	   		>> 用于获取到没有路由到消息队列中的消息。
 * 	   2、ReturnListener 的注意事项 http://www.rabbitmq.com/alarms.html
 * 	   		>> 受到内存和磁盘的限制
 * 	   		>> http://rabbitmq.1065348.n5.nabble.com/ReturnListener-is-not-invoked-td24549.html(一个RabbitMQ disk_free_limit 参数导致ReturnListener没有进入的例子)
 *
 * 	</pre>
 *
 * @author huan.fu
 * @date 2018/8/21 - 15:23
 */
public class RabbitProducer {

	private static final String EXCHANGE_NAME = "exchange_demo";
	private static final String ROUTING_KEY = "missing_routing_key";
	private static final String BINDING_KEY = "bingkey_demo";
	private static final String QUEUE_NAME = "queue_demo";
	private static final String IP_ADDRESS = "140.143.237.224";
	private static final int PORT = 5672;

	public static void main(String[] args) throws IOException, TimeoutException {
		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost(IP_ADDRESS);
		connectionFactory.setPort(PORT);
		connectionFactory.setUsername("root");
		connectionFactory.setPassword("root");

		try (
				// 创建一个连接
				Connection connection = connectionFactory.newConnection();
				// 创建信道
				Channel channel = connection.createChannel()
		) {
			// 创建一个 type="direct"持久化、非自动删除的交换器
			channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, null);
			// 创建一个 持久化、非排他的、非自动删除的交换器
			channel.queueDeclare(QUEUE_NAME, true, false, false, null);
			// 将交换器与队列通过路由键绑定 使用 bindingKey
			channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, BINDING_KEY);

			// 发送一条持久化消息
			String message = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + " 没有被正确路由到消息队列的消息.mandatory参数设置成true";
			try {
				// 使用 routingKey
				channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, true, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));
				System.err.println("消息发送完成......");
			} catch (IOException e) {
				e.printStackTrace();
			}

			/**
			 * 处理生产者没有正确路由到消息队列的消息
			 * 这个可能不会生效：受到 rabbitmq 配置的内存和磁盘的限制 {@link http://www.rabbitmq.com/alarms.html}
			 */
			channel.addReturnListener((replyCode, replyText, exchange, routingKey, properties, body) -> {
				System.out.println("replyCode:" + replyCode);
				System.out.println("replyText:" + replyText);
				System.out.println("exchange:" + exchange);
				System.out.println("routingKey:" + routingKey);
				System.out.println("properties:" + properties);
				System.out.println("body:" + new String(body, StandardCharsets.UTF_8));
			});
		}
	}

}

 

方案二介绍：

     

     使用方案一，我们需要自己写ReturnListener，这样业务代码就变的复杂了，那么有没有一种简单的方法呢？那就是使用 备份交换器(Alternate Exchange)

      声明交换器可以在channel.exchangeDeclare的时候 添加 alternate-exchange 参数来实现，交换器的类型建议声明成 fanout 类型，因为消息被重新发送到备份交换器时的路由键和从生产者出发的路由键是一致的。

示例代码：

public class RabbitProducer {

	private static final String EXCHANGE_NAME = "exchange_demo";
	private static final String BINDING_KEY = "bingkey_demo";
	private static final String QUEUE_NAME = "queue_demo";
	private static final String IP_ADDRESS = "140.143.237.224";
	private static final int PORT = 5672;

	public static void main(String[] args) throws IOException, TimeoutException {
		ConnectionFactory connectionFactory = new ConnectionFactory();
		connectionFactory.setHost(IP_ADDRESS);
		connectionFactory.setPort(PORT);
		connectionFactory.setUsername("root");
		connectionFactory.setPassword("root");

		try (
				// 创建一个连接
				Connection connection = connectionFactory.newConnection();
				// 创建信道
				Channel channel = connection.createChannel()
		) {
			Map<String, Object> arguments = new HashMap<>(16);
			arguments.put("alternate-exchange", "backup-exchange");
			channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, arguments);
			channel.queueDeclare(QUEUE_NAME, true, false, false, null);
			channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, BINDING_KEY);

			// 声明一个 fanout 类型的交换器，建议此处使用 fanout 类型的交换器
			channel.exchangeDeclare("backup-exchange", "fanout", true, false, null);
			// 消息没有被路由的之后存入的队列
			channel.queueDeclare("unRoutingQueue", true, false, false, null);
			channel.queueBind("unRoutingQueue", "backup-exchange", "");

			// 发送一条持久化消息
			String message = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + " 没有被正确的路由到消息队列，此时此消息会进入 unRoutingQueue";
			try {
				// 使用 routingKey
				channel.basicPublish(EXCHANGE_NAME, "not-exists-routing-key", true, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));
				System.err.println("消息发送完成......");
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}
}

 上例图解：