Version: Next

Routing Topic 模式

类似 动态路由

  • Topic 类型的 Exchange 与 Direct 相比，都是可以根据 RoutingKey 把消息路由到不同的队列
  • Topic 类型的 Exchange 可以让队列在绑定 RoutingKey 的时候使用 通配符
  • 这种模式 RoutingKey 一般都是一个或者多个单次组成，多个单词之间用 . 分割，例如 item.insert
    • *：匹配 1 个词
    • #：匹配 1 个或 多 个词
      • a.*：只能匹配 a.b、a.c 之类的
      • a.#：能匹配 a.b.c、a.c.b、a.b.c.d 等

生产者实现

public class ProviderTopic {
public static void main(String[] args) throws IOException {
Connection connection = RabbitMqUtils.getConnection();
Channel channel = connection.createChannel();
String exchangeName = "topics";
// 声明交换机类型为 TOPIC
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC);
// 发送消息
String routingKey = "user.save.1";
channel.basicPublish(exchangeName,
routingKey,
null,
("topic 模式消息 | Routing Key -> [" + routingKey + "]").getBytes());
RabbitMqUtils.closeConnectionAndChannel(connection, channel);
}
}

消费者实现

消费者1
public class ConsumerTopic1 {
public static void main(String[] args) throws IOException {
Connection connection = RabbitMqUtils.getConnection();
Channel channel = connection.createChannel();
String exchangeName = "topics";
channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC);
String queue = channel.queueDeclare().getQueue();
String routingKey = "user.*"; // 通配，2个单次组成，第一个单次必须是 user
channel.queueBind(queue, exchangeName, routingKey);
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 通配规则[" + routingKey + "]-> " + new String(body));
}
});
}
}

测试

消费者1
消费者1 通配规则[user.#]-> topic 模式消息 | Routing Key -> [user.save.1]
消费者2
不消费