在Fanout模式中,一条消息,会被所有订阅的队列都消费。但是,在某些场景下,我们希望不同的
消息被不同的队列给消费。这时就要用到 Direct 类型的Exchange。
在Direct模型下:
- 队列与交换机的绑定,不能是任意绑定了,而是要指定一个RoutingKey(路由 key)
- 消息的发送方在向Exchange发送消息时,也必须指定消息的 RoutingKey。
- Exchange不再把消息交给每一个绑定的队列,而是根据消息的RoutingKey进行判断,只有队列的Routingkey与消息的 Routing key
完全一致
,才会接收到消息。
流程如下图所示:
上图中的解释如下:
- P:生产者,向Exchange发送消息,发送消息时,会指定一个 routing key。
- X:Exchange(交换机)接收生产者的消息,然后把消息递交给与 routing key 完全匹配的队列。
- C1:消费者,其所在队列指定了需要 routing key 为 error 的消息。
- C2:消费者,其所在队列指定了需要 routing key 为 info、error、warning 的消息。
创建生产者
/**
* @author: BNTang
*/
public class Producer {
@Test
public void sendMessage() throws Exception {
Connection connection = RabbitMQUtil.getConnection();
Channel channel = connection.createChannel();
String exchangeName = "logs_direct";
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT);
// 声明一个路由 key
String routingKey = "error";
channel.basicPublish(exchangeName, routingKey, null, ("我是一个直连类型的交换机消息 + routingKey = " + routingKey).getBytes());
System.out.println("消息发送成功");
RabbitMQUtil.closeChannelAndConnection(channel, connection);
}
}
创建消费者 1
/**
* @author BNTang
*/
public class Consumer1 {
@Test
public void receiveMessage() throws Exception {
Connection connection = RabbitMQUtil.getConnection();
Channel channel = connection.createChannel();
String exchangeName = "logs_direct";
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT);
// 得到一个临时队列
String queue = channel.queueDeclare().getQueue();
// 绑定队列到交换机
channel.queueBind(queue, exchangeName, "error");
// 消费消息
channel.basicConsume(queue, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者【1】收到消息" + new String(body));
}
});
System.in.read();
}
}
创建消费者 2
/**
* @author BNTang
*/
public class Consumer2 {
@Test
public void receiveMessage() throws Exception {
Connection connection = RabbitMQUtil.getConnection();
Channel channel = connection.createChannel();
String exchangeName = "logs_direct";
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT);
// 得到一个临时队列
String queue = channel.queueDeclare().getQueue();
// 绑定队列到交换机
channel.queueBind(queue, exchangeName, "info");
channel.queueBind(queue, exchangeName, "error");
channel.queueBind(queue, exchangeName, "warm");
// 消费消息
channel.basicConsume(queue, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者【2】收到消息" + new String(body));
}
});
System.in.read();
}
}
测试
先启动两个消费者,再启动生产者发送 routingKey=error 发现两个都收到如果发送 info 发现只有 consumer2 收到说明它的路由 key 的匹配规则是等值匹配。
更改如上图所示的地方即可进行测试。