一篇文章教你將JAVA的RabbitMQz與SpringBoot整合
本文主要聊SpringBoot整合RabbitMQ,主要分為生產(chǎn)者和消費(fèi)者兩個(gè)工程,目錄結(jié)構(gòu)如下:
先簡(jiǎn)單說(shuō)一下RabbitMQ的一些核心概念:
1.虛擬主機(jī)vhost:vhost是物理隔離的,你可以將vhost看作是一個(gè)個(gè)小型的RabbitMQ
2.交換機(jī)exchange:生產(chǎn)者發(fā)送的消息不是直接到達(dá)隊(duì)列的,而是交換機(jī),然后交換機(jī)再根據(jù)路由key,路由到指定的隊(duì)列,可以理解為一個(gè)分發(fā)消息的角色
3.隊(duì)列:存放消息的地方,消費(fèi)者拿到的消息就是通過(guò)隊(duì)列,可以理解為存放消息的容器
4.bindingKey和routingKey:這兩個(gè)可能比較容易搞混,bindingKey是交換機(jī)和隊(duì)列建立關(guān)系時(shí)候的key,而routingKey則是生產(chǎn)者發(fā)送消息時(shí)候的key。在DirectExchange類型的交換機(jī)中,bindingKey和routingKey是一致的,而在TopicExchange類型的交換機(jī)中,兩者一般是不一致的。
開發(fā)環(huán)境:RabbitMQ:3.7.7、SpringBoot2.1.7
引入相關(guān)依賴:
<dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> <version>2.1.7.RELEASE</version> </dependency> <!-- rabbitmq依賴 --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> <version>2.1.7.RELEASE</version> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <version>2.1.7.RELEASE</version> </dependency> </dependencies>
在生產(chǎn)者和消費(fèi)者工程的application.yml文件添加RabbitMQ配置:
spring: rabbitmq: host: localhost port: 5672 username: guest password: guest
如果是和上面一樣的值,其實(shí)也可以不用配置,因?yàn)镽abbitMQ的默認(rèn)配置就是它們:
接下來(lái),看看幾種交換機(jī)的使用場(chǎng)景,包括:
(1)fanout:發(fā)布訂閱型,可以理解為廣播
(2)direct:直連型,可以理解為點(diǎn)對(duì)點(diǎn)
(3)topic:通配符模式
一、fanout:發(fā)布訂閱型
在生產(chǎn)者工程,添加如下配置:
@Configuration public class FanoutRabbitmqConfig { public static final String EMAIL_QUEUE = "email_fanout"; public static final String SMS_QUEUE = "sms_fanout"; public static final String EXCHANGE_NAME = "exchange_fanout"; // 發(fā)布訂閱模式,不用routingKey @Bean public FanoutExchange getFanoutExchange(){ return new FanoutExchange(EXCHANGE_NAME); } @Bean public Queue getFanoutEmailQueue(){ return new Queue(EMAIL_QUEUE); } @Bean public Queue getFanoutSmsQueue(){ return new Queue(SMS_QUEUE); } // 綁定交換機(jī)和隊(duì)列 @Bean public Binding fanoutEmailExchange(){ return BindingBuilder.bind(getFanoutEmailQueue()).to(getFanoutExchange()); } @Bean public Binding fanoutSmsExchange(){ return BindingBuilder.bind(getFanoutSmsQueue()).to(getFanoutExchange()); } // 用于測(cè)試找到了交換機(jī),但是沒有找到隊(duì)列的情況 @Bean public FanoutExchange notBindingQueue(){ return new FanoutExchange("testNotBindingQueue"); } }
可以看到聲明了名為exchange_fanout的交換機(jī)和兩個(gè)隊(duì)列,并進(jìn)行了綁定,如果向exchange_fanout交換機(jī)發(fā)送消息,則兩個(gè)隊(duì)列都會(huì)收到消息。
二、direct:直連型
在生產(chǎn)者工程添加如下配置:
@Configuration public class RoutingRabbitmqConfig { public static final String EMAIL_QUEUE = "email_routing"; public static final String SMS_QUEUE = "sms_routing"; public static final String EXCHANGE_NAME = "exchange_routing"; public static final String EMAIL_ROUTINGKEY = "inform.email.routingKey"; public static final String SMS_ROUTINGKEY = "inform.sms.routingKey"; @Bean public DirectExchange getDirectExchange(){ return new DirectExchange(EXCHANGE_NAME); } @Bean public Queue getDirectEmailQueue(){ return new Queue(EMAIL_QUEUE); } @Bean public Queue getDirectSmsQueue(){ return new Queue(SMS_QUEUE); } // 綁定交換機(jī)和隊(duì)列 @Bean public Binding directEmailExchange(){ return BindingBuilder.bind(getDirectEmailQueue()).to(getDirectExchange()).with(EMAIL_ROUTINGKEY); } @Bean public Binding directSmsExchange(){ return BindingBuilder.bind(getDirectSmsQueue()).to(getDirectExchange()).with(SMS_ROUTINGKEY); } }
和fanout類型不同的是,這里使用了routingKey進(jìn)行交換機(jī)和隊(duì)列的綁定,如果發(fā)送消息時(shí)的routingKey是inform.sms.routingKey,那么只有sms_routing這個(gè)隊(duì)列可以收到消息,email_routing隊(duì)列則不會(huì)收到消息,這正是直連的含義所在。
三、topic:通配符模式
topic類型的交換機(jī),常使用的通配符是:*和#
(1)星號(hào)*:匹配一個(gè)詞
(2)井號(hào):匹配0個(gè)或多個(gè)詞
詞是通過(guò)點(diǎn)分割的,被點(diǎn)分割的就是一個(gè)詞
先看配置:
@Configuration public class TopicsRabbitmqConfig { public static final String EMAIL_QUEUE = "email_topics"; public static final String SMS_QUEUE = "sms_topics"; public static final String EXCHANGE_NAME = "exchange_topics"; public static final String EMAIL_ROUTINGKEY = "inform.#.email.#"; // * 必須匹配一個(gè)詞,不能是0個(gè) // public static final String EMAIL_ROUTINGKEY = "inform.*.email"; public static final String SMS_ROUTINGKEY = "inform.#.sms.#"; @Bean public TopicExchange getTopicsExchange(){ return new TopicExchange(EXCHANGE_NAME); } @Bean public Queue getTopicsEmailQueue(){ return new Queue(EMAIL_QUEUE); } @Bean public Queue getTopicsSmsQueue(){ return new Queue(SMS_QUEUE); } // 綁定交換機(jī)和隊(duì)列 @Bean public Binding topicsEmailExchange(){ return BindingBuilder.bind(getTopicsEmailQueue()).to(getTopicsExchange()).with(EMAIL_ROUTINGKEY); } @Bean public Binding topicsSmsExchange(){ return BindingBuilder.bind(getTopicsSmsQueue()).to(getTopicsExchange()).with(SMS_ROUTINGKEY); } }
和direct不同的是:bindingKey中可以含有星號(hào)和井號(hào),可以看到:隊(duì)列email_topics的bindingKey(注意,這是bindingKey,不是routingKey)是inform.#.email.#,而隊(duì)列sms_topics則綁定了inform.#.sms.#。如果發(fā)送消息的routingKey是inform開頭,后面含有email,則email_topics隊(duì)列會(huì)收到消息,而sms_topics隊(duì)列不會(huì)收到消息。
topic類型交換機(jī)的強(qiáng)大之處在于,它可以替代前面兩種交換機(jī):
(1)如果bindingKey只有一個(gè)井號(hào),在是fanout
(2)如果bindingKey不含有星號(hào)和井號(hào),則是direct
四、消費(fèi)者端接收消息
消費(fèi)者接收消息配置如下:
@Component public class TopicsConsumer { // topics模式 @RabbitListener(queues = TopicsRabbitmqConfig.EMAIL_QUEUE) public void receiveEmail(String msg){ System.out.println("topics收到email消息: "+msg); } @RabbitListener(queues = TopicsRabbitmqConfig.SMS_QUEUE) public void receiveSms(String msg){ System.out.println("topics收到sms消息: "+msg); } // routing模式 @RabbitListener(queues = RoutingRabbitmqConfig.EMAIL_QUEUE) public void receiveRoutingEmail(String msg){ System.out.println("routing收到email消息: "+msg); } @RabbitListener(queues = RoutingRabbitmqConfig.SMS_QUEUE) public void receiveRoutingSms(String msg){ System.out.println("routing收到sms消息: "+msg); } // 發(fā)布訂閱模式 @RabbitListener(queues = FanoutRabbitmqConfig.EMAIL_QUEUE) public void receiveFanoutEmail(String msg){ System.out.println("fanout收到email消息: "+msg); } @RabbitListener(queues = FanoutRabbitmqConfig.SMS_QUEUE) public void receiveFanoutSms(String msg){ System.out.println("fanout收到sms消息: "+msg); } }
這里主要是使用了RabbitListener注解,指定了接收消息的隊(duì)列。
總結(jié)
本片文章就到這里了,希望能夠給你帶來(lái)幫助,也希望您能夠多多關(guān)注腳本之家的更多內(nèi)容!
相關(guān)文章
如何解決線程太多導(dǎo)致java socket連接池出現(xiàn)的問題
這篇文章主要介紹了如何解決線程太多導(dǎo)致socket連接池出現(xiàn)的問題,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2019-12-12Java中消息隊(duì)列任務(wù)的平滑關(guān)閉詳解
對(duì)于消息隊(duì)列的監(jiān)聽,我們一般使用Java寫一個(gè)獨(dú)立的程序,在Linux服務(wù)器上運(yùn)行。程序啟動(dòng)后,通過(guò)消息隊(duì)列客戶端接收消息,放入一個(gè)線程池進(jìn)行異步處理,并發(fā)的快速處理。這篇文章主要給大家介紹了關(guān)于Java中消息隊(duì)列任務(wù)的平滑關(guān)閉的相關(guān)資料,需要的朋友可以參考下。2017-11-11Java多線程高并發(fā)中的Fork/Join框架機(jī)制詳解
本文主要介紹了 Java 多線程高并發(fā)中的 Fork/Join 框架的基本原理和其使用的工作竊取算法(work-stealing)、設(shè)計(jì)方式和部分實(shí)現(xiàn)源碼,感興趣的朋友跟隨小編一起看看吧2021-11-11java開發(fā)AOP基礎(chǔ)JdkDynamicAopProxy
這篇文章主要為大家介紹了java開發(fā)AOP基礎(chǔ)JdkDynamicAopProxy源碼示例解析,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-07-07springmvc字符編碼過(guò)濾器CharacterEncodingFilter的使用
這篇文章主要介紹了springmvc字符編碼過(guò)濾器CharacterEncodingFilter的使用,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。2021-08-08mybatis 多表關(guān)聯(lián)mapper文件寫法操作
這篇文章主要介紹了mybatis 多表關(guān)聯(lián)mapper文件寫法操作,具有很好的參考價(jià)值,希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2020-12-12Mybatis Mybatis-Plus傳入多個(gè)參數(shù)的處理方式
這篇文章主要介紹了Mybatis Mybatis-Plus傳入多個(gè)參數(shù)的處理方式,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2024-05-05