RabbitMq系列(九)：主题交换Topic Exchange

系列文章

RabbitMq系列(一)：服务器搭建

RabbitMq系列(二)：最简单的例子

RabbitMq系列(三)：工作队列

RabbitMq系列(四)：消息确认和持久性

RabbitMq系列(五)：公平派遣

RabbitMq系列(六)：交换类型以及例子

RabbitMq系列(七)：直接交换Direct exchange

RabbitMq系列(八)：扇出交换Fanout Exchange

RabbitMq系列(九)：主题交换Topic Exchange

RabbitMq系列(十)：标头交换Headers exchange

RabbitMq系列(十一)：远程调用RPC

前言

主题交换是四个主要交换模式中，变化最多、最灵活的交换模式。主题交换的消息传递主要依赖于路由键，路由键不同，交换机匹配的方式不同。

主题交换的主要关注点在路由键，路由键通常是由零个或者多个有意义的单词通过点号( . )分隔拼接而成，类似于： topic.route.one ，topic.route，topic 等等，路由键最多只能有255个字节。

主题交换中一般的路由键规则跟直接交换路由规则大致相同，都是直接比较是否相等，但是主题交换有特殊的路由键规则。

主题交换中有个两个特殊的匹配符号：
* ：匹配任意一个单词

# ：匹配零个或者多个单词

不带两个特殊符号的路由键匹配规则的同直接交换匹配规则一样，带两个特殊符号的类似于模糊匹配，只带单个 # 的就是扇出交换啦，带特殊符号的路由键类似于： topic.#.#* ， topic.route.#.# ，topic.route.*# ， topic.route.one ， # ，* 等等

实际演示

主题交换作为四个主要交换中最灵活的交换，路由键的定义非常丰富，所以这里采用三个消费者和一个生产者来进行测试，路由键从简单的复杂。

示例1

环境

身份	交换机	队列	routeKey
消费者1	topic.exchange	amq.gen-7V2hbBS1AIyJkSOSCuWdRA	topic.route.two
消费者2	topic.exchange	amq.gen-P4DJ32qk9nvX3UE7PM4T_A	topic.route.*
消费者3	topic.exchange	amq.gen-X_JrOM4KZP_KvMk3IcaH1g	topic.#
生产者	topic.exchange	无	topic.route.one

消费者1

/*** 消费者1* @Tag 主题交换Topic exchange*/
public class MQConsumerOne {public static void main(String[] args) {try {consumerMsg("topic.exchange","topic.route.two");} catch (Exception e) {e.printStackTrace();}}public static void consumerMsg(String exchangeName,String routeKey) throws IOException, TimeoutException {//创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();//设置连接地址connectionFactory.setHost("10.0.10.3");//设置连接端口connectionFactory.setPort(5672);//设置连接的虚拟机connectionFactory.setVirtualHost("mqtest");//设置连接用户connectionFactory.setUsername("mqtest");//设置连接用户密码connectionFactory.setPassword("test123");//创建连接Connection connection = connectionFactory.newConnection();//创建通道Channel channel = connection.createChannel();//声明交换机channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC);//声明队列(临时排它队列)String queueName = channel.queueDeclare().getQueue();//绑定交换机channel.queueBind(queueName,exchangeName,routeKey);//消费消息（主动确认）channel.basicConsume(queueName,true,(tag,msg)->{System.out.println(new String(msg.getBody(),"UTF-8"));},(cancel)->{});/* //关闭连接channel.close();connection.close();*/}
}

消费者2

/*** 消费者2* @Tag 主题交换Topic exchange*/
public class MQConsumerTwo {public static void main(String[] args) {try {consumerMsg("topic.exchange","topic.route.*");} catch (Exception e) {e.printStackTrace();}}public static void consumerMsg(String exchangeName,String routeKey) throws IOException, TimeoutException {//创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();//设置连接地址connectionFactory.setHost("10.0.10.3");//设置连接端口connectionFactory.setPort(5672);//设置连接的虚拟机connectionFactory.setVirtualHost("mqtest");//设置连接用户connectionFactory.setUsername("mqtest");//设置连接用户密码connectionFactory.setPassword("test123");//创建连接Connection connection = connectionFactory.newConnection();//创建通道Channel channel = connection.createChannel();//声明交换机channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC);//声明队列(临时排它队列)String queueName = channel.queueDeclare().getQueue();//绑定交换机channel.queueBind(queueName,exchangeName,routeKey);//消费消息（主动确认）channel.basicConsume(queueName,true,(tag,msg)->{System.out.println(new String(msg.getBody(),"UTF-8"));},(cancel)->{});/* //关闭连接channel.close();connection.close();*/}
}

消费者3

/*** 消费者3* @Tag 主题交换Topic exchange*/
public class MQConsumerThree {public static void main(String[] args) {try {consumerMsg("topic.exchange","topic.#");} catch (Exception e) {e.printStackTrace();}}public static void consumerMsg(String exchangeName,String routeKey) throws IOException, TimeoutException {//创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();//设置连接地址connectionFactory.setHost("10.0.10.3");//设置连接端口connectionFactory.setPort(5672);//设置连接的虚拟机connectionFactory.setVirtualHost("mqtest");//设置连接用户connectionFactory.setUsername("mqtest");//设置连接用户密码connectionFactory.setPassword("test123");//创建连接Connection connection = connectionFactory.newConnection();//创建通道Channel channel = connection.createChannel();//声明交换机channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC);//声明队列(临时排它队列)String queueName = channel.queueDeclare().getQueue();//绑定交换机channel.queueBind(queueName,exchangeName,routeKey);//消费消息（主动确认）channel.basicConsume(queueName,true,(tag,msg)->{System.out.println(new String(msg.getBody(),"UTF-8"));},(cancel)->{});/* //关闭连接channel.close();connection.close();*/}
}

唯一的生产者

/*** 生产者* @Tag 主题交换Topic exchange*/
public class MQProducer {public static void main(String[] args) {try {producerMsg("topic.exchange","topic.route.one","想要我的消息？就看你有没得这个本事！");} catch (Exception e) {e.printStackTrace();}}public static void producerMsg(String exchangeName,String routeKey,String msg) throws IOException, TimeoutException {//创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();//设置连接地址connectionFactory.setHost("10.0.10.3");//设置连接端口connectionFactory.setPort(5672);//设置连接的虚拟机connectionFactory.setVirtualHost("mqtest");//设置连接用户connectionFactory.setUsername("mqtest");//设置连接用户密码connectionFactory.setPassword("test123");//创建连接Connection connection = connectionFactory.newConnection();//创建通道Channel channel = connection.createChannel();//声明临时交换机channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC,false);//创建生产者channel.basicPublish(exchangeName,routeKey,null,msg.getBytes("UTF-8"));//关闭连接channel.close();connection.close();}
}

完成代码逻辑过后，分别启动消费者1、消费者2、消费者3和唯一的生产者，观察消息的收发情况。这里消费者2、消费者3收到消息，消费者1未收到消息。

分析一下：

1) 消费者1的 routeKey ： topic.route.two

2) 消费者2的 routeKey ： topic.route.*

3）消费者3的 routeKey ： topic.#

4）唯一生产者的 routeKey ： topic.route.one

咱们一个一个来，往下看