Routing Topic
Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key的时候使用通配符！这种模型Routingkey一般都是由一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert、 item.#、item.*。
统配符
* (star) can substitute for exactly one word. 匹配不多不少恰好1个词 # (hash) can substitute for zero or more words. 匹配[0-n]个词 a.# 可以匹配 a.b、a.b.c、a.b.c.d 等只要是a.开头的情况 a.* 只能匹配 a.b 这种后面只有一个单词的情况
通配符可以出现在
注意：RoutingKey的任意位置。
创建生产者
public class MyProducer { @Test public void test() throws Exception { // 交换机 String exchange = "logs_topic"; // 创建工厂 ConnectionFactory factory = new ConnectionFactory(); factory.setVirtualHost("/"); factory.setHost("xuewei.world"); factory.setUsername("xuewei"); factory.setPassword("123456"); factory.setPort(5672); // 创建连接和通道 Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); // 声明交换机 channel.exchangeDeclare(exchange, "topic"); // 发布消息 channel.basicPublish(exchange, "a.b", null, "a.b".getBytes()); channel.basicPublish(exchange, "a.b.c", null, "a.b.c".getBytes()); channel.basicPublish(exchange, "a.b.c.d", null, "a.b.c.d" .getBytes()); channel.basicPublish(exchange, "a.b.c.d.e", null, "a.b.c.d.e".getBytes()); } }
创建消费者1
public class MyConsumer1 { public static void main(String[] args) throws Exception { // 指定交换机 String exchange = "logs_topic"; // 创建工厂 ConnectionFactory factory = new ConnectionFactory(); factory.setVirtualHost("/"); factory.setHost("xuewei.world"); factory.setUsername("xuewei"); factory.setPassword("123456"); factory.setPort(5672); // 创建连接和通道 Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); // 绑定交换机 channel.exchangeDeclare(exchange, "topic"); // 创建临时队列 String queue = channel.queueDeclare().getQueue(); // 将临时队列绑定exchange channel.queueBind(queue, exchange, "a.*"); channel.queueBind(queue, exchange, "#.d.#"); // 处理消息 channel.basicConsume(queue, true, 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)); // TODO 业务处理 } }); } }
创建消费者2
public class MyConsumer2 { public static void main(String[] args) throws Exception { // 指定交换机 String exchange = "logs_topic"; // 创建工厂 ConnectionFactory factory = new ConnectionFactory(); factory.setVirtualHost("/"); factory.setHost("xuewei.world"); factory.setUsername("xuewei"); factory.setPassword("123456"); factory.setPort(5672); // 创建连接和通道 Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); // 绑定交换机 channel.exchangeDeclare(exchange, "topic"); // 创建临时队列 String queue = channel.queueDeclare().getQueue(); // 将临时队列绑定exchange channel.queueBind(queue, exchange, "#.b.#"); // 处理消息 channel.basicConsume(queue, true, 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)); // TODO 业务处理 } }); } }
生产者生产的消息：a.b、a.b.c、a.b.c.d、a.b.c.d.e
消费者1接受的消息规则为：
channel.queueBind(queue, exchange, "a.*"); channel.queueBind(queue, exchange, "#.d.#");
所以消费者1将会接收到：a.b、a.b.c.d、a.b.c.d.e
消费者2接受的消息规则为：
channel.queueBind(queue, exchange, "#.b.#");
所以消费者2将会接收到：a.b、a.b.c、a.b.c.d、a.b.c.d.e
