spring boot使用RabbitMQ实现topic 主题
作者:阿波罗程序猿
前一篇我们实现了消息系统的灵活配置。代替了使用扇形(fanout)交换器的配置。使用直连(direct)交换器,并且基于路由键后可以有选择性接收消息的能力。
虽然使用直连交换器可以改善我们的系统,但是它仍有局限性,它不能实现多重条件的路由。
在我们的消息系统中,我们不仅想要订阅基于路由键的队列,还想订阅基于生产消息的源。这些概念来自于Unix工具syslog。该日志基于严格的(info/warn/crit...) 和容易的(auth/cron/kern...)的路由方式。我们的例子比这个要简单。
这个例子将会给我们很大的灵活性,比如我们既想监听来‘cron'自错误的日志又想监听来自‘kern'的所有日志。
为了实现这个灵活性,我们需要知道更多关于主题交换器的内容。
主题交换器
使用主题交换器时不能采用任意写法的路由键,路由键的形式应该是由点分割的单词。用什么词都行,通常都是能表明意义的。例如"stock.usd.nyse", "nyse.vmw", "quick.orange.rabbit"。但字数大小被限制在最多255字节。
使用主题交换器定义路由键需要注意点2点
- *星号代表一个单词。
- #井号代表0个或多个单词。
定义符合主题交换器的路由键
在这个例子中,我们将发送所有描述动物的消息。这个消息将会和由3个单词2个点构成的路由键一起发送。第一个单词是表述速度,第二个描述颜色,第三个描述种类:"<speed>.<colour>.<species>"。
创建三种绑定,Q1和键"*.orange.*"绑定,Q2和"*.*.rabbit" 、"lazy.#"绑定。
三种绑定关系的概述为:
- Q1 对橙色的动物感兴趣。(队列1)
- Q2 对所有关于兔子和所有关于慢速的动物感兴趣。(队列2)
一个和路由键被设置成"quick.orange.rabbit"的消息将会被传递到Q1、Q2这两个队列中。"lazy.orange.elephant" 也会这样。"quick.orange.fox"会去第一个队列,"lazy.brown.fox"会去第二个队列,"lazy.pink.rabbit"会去第二个队列及时它匹配了2次绑定。"quick.brown.fox"因为不匹配哪也去不了,会被丢弃。
那么像"orange" 、 "quick.orange.male.rabbit"这样的呢?因为没有匹配到任何绑定也会被丢弃。
那么像"lazy.orange.male.rabbit"也是四个词的路由键呢?,由于匹配到了lazy.#这个将会被传递到第二个队列中。
主题交换器的小技巧
主题交换器是牛逼的并且表现的与其它交换器相似。
- 当一个队列和 "#" 绑定键绑定时,该队列能收到所有的消息,这点与扇形(fanout)交换器类似。
- 当不使用 "*" and "#" 时,主题交换器就与直连交换器没啥两样。
代码示例
代码与之前的路由代码没啥两样,请看
Config.java
package com.zb.rabbitMQtest.t5topics.config; import org.springframework.amqp.core.*; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; /** * @author 张博 */ @Configuration(value = "t5Config") public class Config { /** * 创建人:张博 * 时间:2018/3/5 上午10:45 * @apiNote 定义主题交换器 */ @Bean public TopicExchange topicExchange() { return new TopicExchange("topic-exchange"); } /** * 创建人:张博 * 时间:2018/3/5 上午10:48 * @apiNote 定义自动删除匿名队列 */ @Bean public Queue autoDeleteQueue0() { return new AnonymousQueue(); } /** * 创建人:张博 * 时间:2018/3/5 上午10:48 * @apiNote 定义自动删除匿名队列 */ @Bean public Queue autoDeleteQueue1() { return new AnonymousQueue(); } /** * 创建人:张博 * 时间:2018/3/5 上午10:48 * @param topicExchange 主题交换器 * @param autoDeleteQueue0 自动删除队列 * @apiNote 绑定使用路由键为 orange 的 autoDeleteQueue0 队列到主题交换器上 * @return Binding */ @Bean public Binding binding0a(TopicExchange topicExchange, Queue autoDeleteQueue0) { return BindingBuilder.bind(autoDeleteQueue0).to(topicExchange).with("*.orange.*"); } /** * 创建人:张博 * 时间:2018/3/5 上午10:48 * @param topicExchange 主题交换器 * @param autoDeleteQueue1 自动删除队列 * @apiNote 绑定使用路由键为 black 的 autoDeleteQueue1 队列到主题交换器上 * @return Binding */ @Bean public Binding binding1a(TopicExchange topicExchange, Queue autoDeleteQueue1) { return BindingBuilder.bind(autoDeleteQueue1).to(topicExchange).with("*.*.rabbit"); } /** * 创建人:张博 * 时间:2018/3/5 上午10:48 * @param topicExchange 主题交换器 * @param autoDeleteQueue1 自动删除队列 * @apiNote 绑定使用路由键为 green 的 autoDeleteQueue1 队列到主题交换器上 * @return Binding */ @Bean public Binding binding1b(TopicExchange topicExchange, Queue autoDeleteQueue1) { return BindingBuilder.bind(autoDeleteQueue1).to(topicExchange).with("lazy.#"); } }
Receiver.java
package com.zb.rabbitMQtest.t5topics.receiver; import org.springframework.amqp.rabbit.annotation.RabbitListener; import org.springframework.stereotype.Component; /** * @author 张博 */ @Component(value = "t5Receiver") public class Receiver { @RabbitListener(queues = "#{autoDeleteQueue0.name}") public void receiver0(String str) { System.out.println("receiver0++++++++++:" + str); //try { // Thread.sleep(1000); //} catch (InterruptedException e) { // e.printStackTrace(); //} } @RabbitListener(queues = "#{autoDeleteQueue1.name}") public void receiver1(String str) { System.out.println("receiver1++++++++++:" + str); //try { // Thread.sleep(1000); //} catch (InterruptedException e) { // e.printStackTrace(); //} } }
Send.java
package com.zb.rabbitMQtest.t5topics.send; import org.springframework.amqp.core.TopicExchange; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; /** * @author 张博 */ @Component(value = "t5Send") public class Send { @Autowired private TopicExchange topicExchange; @Autowired private RabbitTemplate rabbitTemplate; private String[] keys = {"quick.orange.rabbit", "lazy.orange.elephant", "quick.orange.fox", "lazy.brown.fox", "lazy.pink.rabbit", "quick.brown.fox"}; public void send() { String message = "哈哈哈"; for (int i = 0; i < 5; i++) { System.out.println("send++++++++++:".concat(message)); rabbitTemplate.convertAndSend(topicExchange.getName(), keys[5], message); } } }
SendTest.java
package com.zb.rabbitMQtest.t5topics.send; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.context.SpringBootTest; import org.springframework.test.context.junit4.SpringRunner; /** * @author 张博 */ @RunWith(SpringRunner.class) @SpringBootTest public class SendTest { @Autowired private Send send; @Test public void send() throws Exception { send.send(); } }
测试结果我就不放了 大家请自行查看。
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持脚本之家。