在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 的匹配规则是等值匹配。
更改如上图所示的地方即可进行测试。