快捷導(dǎo)航

Spring Boot整合RabbitMQ實(shí)例（Topic模式）

更新時(shí)間：2017年04月26日 11:53:51 作者：牛頭人

Topic Exchange 轉(zhuǎn)發(fā)消息主要是根據(jù)通配符。接下來(lái)通過(guò)本文給大家分享Spring Boot整合RabbitMQ實(shí)例（Topic模式），需要的朋友參考下吧

1.Topic交換器介紹

Topic Exchange 轉(zhuǎn)發(fā)消息主要是根據(jù)通配符。在這種交換機(jī)下，隊(duì)列和交換機(jī)的綁定會(huì)定義一種路由模式，那么，通配符就要在這種路由模式和路由鍵之間匹配后交換機(jī)才能轉(zhuǎn)發(fā)消息。

在這種交換機(jī)模式下：

路由鍵必須是一串字符，用句號(hào)（.）隔開(kāi)，比如說(shuō) agreements.us，或者 agreements.eu.stockholm 等。

路由模式必須包含一個(gè) 星號(hào)（*），主要用于匹配路由鍵指定位置的一個(gè)單詞，比如說(shuō)，一個(gè)路由模式是這樣子：agreements..b.*，那么就只能匹配路由鍵是這樣子的：第一個(gè)單詞是 agreements，第四個(gè)單詞是 b。井號(hào)（#）就表示相當(dāng)于一個(gè)或者多個(gè)單詞，例如一個(gè)匹配模式是agreements.eu.berlin.#，那么，以agreements.eu.berlin開(kāi)頭的路由鍵都是可以的。
具體代碼發(fā)送的時(shí)候還是一樣，第一個(gè)參數(shù)表示交換機(jī)，第二個(gè)參數(shù)表示routing key，第三個(gè)參數(shù)即消息。如下：

rabbitTemplate.convertAndSend("testTopicExchange","key1.a.c.key2", " this is RabbitMQ!");

topic 和 direct 類似, 只是匹配上支持了"模式", 在"點(diǎn)分"的 routing_key 形式中, 可以使用兩個(gè)通配符:

*表示一個(gè)詞.

#表示零個(gè)或多個(gè)詞.

如上圖所示:此類交換器使得來(lái)自不同的源頭的消息可以到達(dá)一個(gè)對(duì)列，其實(shí)說(shuō)的更明白一點(diǎn)就是模糊匹配的意思，例如：上圖中紅色對(duì)列的routekey為usa.#，#代表匹配任意字符，但是要想消息能到達(dá)此對(duì)列，usa.必須匹配后面的#好可以隨意。圖中usa.news,usa.weather都能找到紅色隊(duì)列，符號(hào)“#”匹配一個(gè)或多個(gè)詞，符號(hào)“”匹配不多不少一個(gè)詞。因此“usa.#”能夠匹配到“usa.news.XXX”，但是“usa.” 只會(huì)匹配到“usa.XXX”。
注：交換器說(shuō)到底是一個(gè)名稱與隊(duì)列綁定的列表。當(dāng)消息發(fā)布到交換器時(shí)，實(shí)際上是由你所連接的信道，將消息路由鍵同交換器上綁定的列表進(jìn)行比較，最后路由消息

2.示例代碼

1）.RabbitMQ的Topic的bean配置

RabbitTopic.java類：

package com.example.rabbitmqtopic;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitTopic {
 final static String message = "topic.message";
 final static String messages = "topic.messages";
 //創(chuàng)建隊(duì)列
 @Bean
 public Queue queueMessage() {
  return new Queue(RabbitTopic.message);
 }
 //創(chuàng)建隊(duì)列
 @Bean
 public Queue queueMessages() {
  return new Queue(RabbitTopic.messages);
 }
 //創(chuàng)建交換器
 @Bean
 TopicExchange exchange() {
  return new TopicExchange("topicExchange");
 }
　 //對(duì)列綁定并關(guān)聯(lián)到ROUTINGKEY
 @Bean
 Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
  return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
 }
 //對(duì)列綁定并關(guān)聯(lián)到ROUTINGKEY
 @Bean
 Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
  return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");//*表示一個(gè)詞,#表示零個(gè)或多個(gè)詞
　　}
}

2）.消息生產(chǎn)者生產(chǎn)消息

TopicSender.java類：

package com.example.rabbitmqtopic.rabbitmq;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Component
public class TopicSender {
 @Autowired
 private AmqpTemplate rabbitTemplate;
 public void send() {
  String context = "hi, i am message all";
  System.out.println("Sender : " + context);
  this.rabbitTemplate.convertAndSend("topicExchange", "topic.1", context);
 }
 public void send1() {
  String context = "hi, i am message 1";
  System.out.println("Sender : " + context);
  this.rabbitTemplate.convertAndSend("topicExchange", "topic.message", context);
 }
 public void send2() {
  String context = "hi, i am messages 2";
  System.out.println("Sender : " + context);
  this.rabbitTemplate.convertAndSend("topicExchange", "topic.messages", context);
　　}
}

3）.消息消費(fèi)者

TopicReceiver.java類：

package com.example.rabbitmqtopic.rabbitmq;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
@RabbitListener(queues = "topic.message")
public class TopicReceiver {
 @RabbitHandler
 public void process(String message) {
  System.out.println("Topic Receiver1 : " + message);
 }
}

TopicReceiver2.java類：

package com.example.rabbitmqtopic.rabbitmq;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
@Component
@RabbitListener(queues = "topic.messages")
public class TopicReceiver2 {
 @RabbitHandler
 public void process(String message) {
  System.out.println("Topic Receiver2 : " + message);
 }
}

4）.測(cè)試

RabbitMQTopicTest.java類：

package com.example.rabbitmqtopic.rabbitmq;
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;
@RunWith(SpringRunner.class)
@SpringBootTest
public class RabbitMQTopicTest {
 @Autowired
 private TopicSender sender;
 @Test
 public void topic() throws Exception {
  sender.send();
 }
 @Test
 public void topic1() throws Exception {
  sender.send1();
 }
 @Test
 public void topic2() throws Exception {
  sender.send2();
 }
}

以上所述是小編給大家介紹的Spring Boot整合RabbitMQ實(shí)例（Topic模式），希望對(duì)大家有所幫助，如果大家有任何疑問(wèn)請(qǐng)給我留言，小編會(huì)及時(shí)回復(fù)大家的。在此也非常感謝大家對(duì)腳本之家網(wǎng)站的支持！

您可能感興趣的文章: