概述
Topic 类型的Exchange与Direct相比，都是可以根据 RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！这种模式Routingkey 一般都是由一个或多个单词组成，多个单词之间以 ”.” 分割，例如： item.insert
- # 统配符
- * (star) can substitute for exactly one word：匹配不多不少恰好1个词
- # (hash) can substitute for zero or more words：匹配0个或多个词
# 示例如下所示：
audit.#：匹配 audit.irs.corporate 或者 audit.irs 等
audit.*：只能匹配 audit.irs
创建生产者
/**
* @author: BNTang
*/
public class Producer {
@Test
public void sendMessage() throws Exception {
Connection connection = RabbitMQUtil.getConnection();
Channel channel = connection.createChannel();
String exchangeName = "topic";
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC);
// 声明一个路由 key
String routingKey = "user";
channel.basicPublish(exchangeName, routingKey, null, ("我是一个topic类型的交换机消息 + 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 = "topic";
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC);
// 得到一个临时队列
String queue = channel.queueDeclare().getQueue();
// 绑定队列到交换机
channel.queueBind(queue, exchangeName, "user.*");
// 消费消息
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 = "topic";
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC);
// 得到一个临时队列
String queue = channel.queueDeclare().getQueue();
// 绑定队列到交换机
channel.queueBind(queue, exchangeName, "user.#");
// 消费消息
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();
}
}
测试
先启动消费者1和消费者2，再分别发送routingKey=user和user.add和user.add.insert等类型的消息。查看结果即可。
