共计 1825 个字符,预计需要花费 5 分钟才能阅读完成。
当消息经过交换器准备路由给队列的时候,发现没有对应的队列可以投递信息,在 rabbitmq 中会默认丢弃消息,如果我们想要监测哪些消息被投递到没有对应的队列,我们可以用备用交换器来实现。
大概原理如下,如下图所示,消息发送给交换器,交换器发现没有可路由的队列,于是消息发给备用交换器,备用交换器再发给队列 2,由队列 2 的消费者来处理消息。
示例
我们在 RabbitMQ – 交换器知道,交换器的定义,需要一个参数,可以通过参数的方式,来指定备用交换器。参数的 key 是 alternate-exchange
,value 是交换器的名称。通常备用交换器的类型是fanout
。
生产者中,定义一个交换器和备用交换器,此时并没有响应路由的队列。
public static void main(String[] args) throws IOException, TimeoutException {
// 声明一个连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 创建一个与 rabbitmq 服务器的连接
// 创建一个 Channel
try (Connection connection = factory.newConnection();
Channel channel = connection.createChannel()) {
// 备用交换器
channel.exchangeDeclare("back.exchange", BuiltinExchangeType.FANOUT);
Map<String, Object> map = new HashMap<>();
map.put("alternate-exchange", "back.exchange");
// 定义交换器
channel.exchangeDeclare("exchange", BuiltinExchangeType.DIRECT, false, false, map);
String[] routingKeys = {"exchange.no.found"};
for (int i = 0; i < routingKeys.length; i++) {
// 把消息发送到队列中
channel.basicPublish("exchange", routingKeys[i], null, routingKeys[i].getBytes());
}
System.out.println("Sent complete");
}
}
消费者,订阅的是备用交换器信息
// 定义队列的名称
public final static String QUEUE_NAME = "back.queue";
public static void main(String[] args) throws IOException, TimeoutException {
// 声明一个连接工厂
ConnectionFactory factory = new ConnectionFactory();
// 创建一个与 rabbitmq 服务器的连接
Connection connection = factory.newConnection();
// 创建一个 Channel
Channel channel = connection.createChannel();
// 通过 Channel 定义队列
// 这边声明队列,是因为可能在发布服务器之前启动消费者,所以在尝试使用队列中的消息之前确保队列存在
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 绑定交换器、路由键、队列
channel.queueBind(QUEUE_NAME, "back.exchange", "");
System.out.println("Waiting for messages.");
// 异步回调处理
DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");
System.out.println("back.queue Received'" + message + "'");
};
// 接收消息
channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {});
}
运行结果如下,收到了备用交换器发送来的消息
正文完