RabbitMQ：交换机（topic exchange）_综合

首先看看AMQP协议，对RabbitMQ的架构会更了解。

深入理解AMQP协议

创建一个Maven项目，根据自己服务器RabbitMQ的版本导入相应的包。
在这里插入图片描述

		<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>3.6.5</version></dependency>

主题交换机

前面提到的 direct exchange的路由规则是严格意义上的匹配，换言之 Routing Key 必须与 Binding Key 相匹配的时候才将消息传送给 Queue。

而topic exchange的路由规则是一种模糊匹配，可以通过通配符满足一部分规则就可以传送。

它的约定是：

1）binding key 中可以存在两种特殊字符 “*” 与“#”，用于做模糊匹配，其中 “*” 用于匹配一个单词，“#”用于匹配多个单词（可以是零个）。

2）routing key 为一个句点号 “.” 分隔的字符串（我们将被句号 “. ” 分隔开的每一段独立的字符串称为一个单词），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”，binding key 与 routing key 一样也是句点号 “.” 分隔的字符串。

在这里插入图片描述
当生产者发送消息 Routing Key=F.C.E 的时候，这时候只满足 Queue1，所以会被路由到 Queue1 中，如果 Routing Key=A.C.E 这时候会被同时路由到 Queue1 和 Queue2 中，如果 Routing Key=A.F.B 时，这里只会发送一条消息到 Queue2 中。

主题交换机拥有非常广泛的用户案例。无论何时，当一个问题涉及到那些想要有针对性的选择需要接收消息的多消费者 / 多应用（multiple consumers/applications）的时候，主题交换机都可以被列入考虑范围。

生产端

package com.kaven.rabbitmq.exchange.topicExchange;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class TopicProducer {
    // 自己服务器的IPprivate static String ip = "IP";// RabbitMQ启动的默认端口，也是应用程序进行连接RabbitMQ的端口private static int port = 5672;// RabbitMQ有一个 "/" 的虚拟主机private static String virtualHost = "/";// topic exchange ，RabbitMQ提供的topic exchangeprivate static String exchangeName = "amq.topic";// exchange typeprivate static String exchangeType= "topic";// 交换机路由的routingKeyprivate static String[] routingKey = {
    "test.kaven.wyy" , "test" , "test.topic"};public static void main(String[] args) throws IOException, TimeoutException {
    // 1 创建一个连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(ip);connectionFactory.setPort(port);connectionFactory.setVirtualHost(virtualHost);// 2 创建连接Connection connection = connectionFactory.newConnection();// 3 创建ChannelChannel channel = connection.createChannel();// 4 发送消息for (int i = 0; i < routingKey.length; i++) {
    String msg = "RabbitMQ：Topic Exchange 发送数据 , routingKey："+routingKey[i];channel.basicPublish(exchangeName ,routingKey[i] ,null, msg.getBytes());}// 5 关闭连接channel.close();connection.close();}
}

消费端

这里创建两个消费者，为每个消费者创建一个线程，以便监听生产端发送过来的消息。

package com.kaven.rabbitmq.exchange.topicExchange;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class TopicConsumer {
    // 自己服务器的IPprivate static String ip = "IP";// RabbitMQ启动的默认端口，也是应用程序进行连接RabbitMQ的端口private static int port = 5672;// RabbitMQ有一个 "/" 的虚拟主机private static String virtualHost = "/";// topic exchange ，RabbitMQ提供的topic exchangeprivate static String exchangeName = "amq.topic";// exchange typeprivate static String exchangeType= "topic";// 队列名private static String[] queueName = {
    "queue_#" , "queue_*"};// 队列与交换机绑定的routingKeyprivate static String[] routingKey = {
    "test.#" , "test.*"};public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
    // 1 创建一个连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(ip);connectionFactory.setPort(port);connectionFactory.setVirtualHost(virtualHost);// 2 创建连接Connection connection = connectionFactory.newConnection();// 3 创建ChannelChannel channel = connection.createChannel();for (int i = 0; i < queueName.length; i++) {
    // 4 定义Queue ，将Queue绑定到direct exchangechannel.queueDeclare(queueName[i],true , false , false , null);channel.queueBind(queueName[i] , exchangeName , routingKey[i]);}// 5 创建消费者QueueingConsumer consumer0 = new QueueingConsumer(channel);QueueingConsumer consumer1 = new QueueingConsumer(channel);// 6 设置channel.basicConsume(queueName[0] , true , consumer0);channel.basicConsume(queueName[1] , true , consumer1);// 7 接收消息Thread thread0 = new Thread(new MyRunnable(consumer0 , routingKey[0]));Thread thread1 = new Thread(new MyRunnable(consumer1 , routingKey[1]));thread0.start();thread1.start();thread0.join();thread1.join();}
}

package com.kaven.rabbitmq.exchange.topicExchange;import com.rabbitmq.client.QueueingConsumer;public class MyRunnable implements Runnable {
    private QueueingConsumer consumer;private String routingKey;public MyRunnable(QueueingConsumer consumer , String routingKey) {
    this.consumer = consumer;this.routingKey = routingKey;}@Overridepublic void run() {
    while(true){
    try {
    QueueingConsumer.Delivery delivery = consumer.nextDelivery();String msg = new String(delivery.getBody());System.out.println(routingKey+"[收到]-"+msg);} catch (InterruptedException e) {
    e.printStackTrace();}}}
}

测试

因为这里使用的是RabbitMQ提供给我们的topic exchange，所以我们无需自己定义。
因为交换机已经定义好了，所以无论先启动生产端还是消费端，消费端都可以成功收到消息。
消费端输出如下：

test.#[收到]-RabbitMQ：Topic Exchange 发送数据 , routingKey：test.kaven.wyy
test.*[收到]-RabbitMQ：Topic Exchange 发送数据 , routingKey：test.topic
test.#[收到]-RabbitMQ：Topic Exchange 发送数据 , routingKey：test
test.#[收到]-RabbitMQ：Topic Exchange 发送数据 , routingKey：test.topic

很明显输出结果符合上面对topic exchange路由规则的解释。

看看RabbitMQ Management。
在这里插入图片描述