怎么在spring中使用RabbitMQ傳遞消息?針對(duì)這個(gè)問(wèn)題,這篇文章詳細(xì)介紹了相對(duì)應(yīng)的分析和解答,希望可以幫助更多想解決這個(gè)問(wèn)題的小伙伴找到更簡(jiǎn)單易行的方法。
為企業(yè)提供網(wǎng)站建設(shè)、網(wǎng)站制作、網(wǎng)站優(yōu)化、全網(wǎng)營(yíng)銷推廣、競(jìng)價(jià)托管、品牌運(yùn)營(yíng)等營(yíng)銷獲客服務(wù)。成都創(chuàng)新互聯(lián)擁有網(wǎng)絡(luò)營(yíng)銷運(yùn)營(yíng)團(tuán)隊(duì),以豐富的互聯(lián)網(wǎng)營(yíng)銷經(jīng)驗(yàn)助力企業(yè)精準(zhǔn)獲客,真正落地解決中小企業(yè)營(yíng)銷獲客難題,做到“讓獲客更簡(jiǎn)單”。自創(chuàng)立至今,成功用技術(shù)實(shí)力解決了企業(yè)“網(wǎng)站建設(shè)、網(wǎng)絡(luò)品牌塑造、網(wǎng)絡(luò)營(yíng)銷”三大難題,同時(shí)降低了營(yíng)銷成本,提高了有效客戶轉(zhuǎn)化率,獲得了眾多企業(yè)客戶的高度認(rèn)可!
構(gòu)建
構(gòu)建一個(gè)使用Spring AMQP發(fā)布消息的應(yīng)用程序,RabbitTemplate并使用POJO訂閱消息MessageListenerAdapter。
創(chuàng)建Rabbit MQ消息接收器
使用任何基于消息傳遞的應(yīng)用程序,您需要?jiǎng)?chuàng)建一個(gè)響應(yīng)已發(fā)布消息的接收器。
@Slf4j @Component public class Receiver { private CountDownLatch latch = new CountDownLatch(1); public void receiveMessage(String message){ log.info("Received < " + message + " >"); latch.countDown(); } public CountDownLatch getLatch(){ return latch; } }
Receiver是一個(gè)簡(jiǎn)單的POJO,它定義了一種接收消息的方法。當(dāng)您注冊(cè)它以接收消息時(shí),您可以將其命名為任何您想要的名稱。
為方便起見,這個(gè)POJO也有一個(gè)CountDownLatch。這允許它發(fā)信號(hào)通知接收到消息。這是您不太可能在生產(chǎn)應(yīng)用程序中實(shí)現(xiàn)的。
注冊(cè)監(jiān)聽器并發(fā)送消息
Spring AMQP RabbitTemplate 提供了使用RabbitMQ發(fā)送和接收消息所需的一切。具體來(lái)說(shuō),你需要配置:
消息偵聽器容器
聲明隊(duì)列,交換以及它們之間的綁定
用于發(fā)送一些消息以測(cè)試偵聽器的組件
Spring Boot會(huì)自動(dòng)創(chuàng)建連接工廠和RabbitTemplate,從而減少您必須編寫的代碼量。
您將使用RabbitTemplate發(fā)送消息,并將Receiver使用消息偵聽器容器注冊(cè),以接收消息。連接工廠驅(qū)動(dòng)兩者,允許它們連接到RabbitMQ服務(wù)器。
@SpringBootApplication public class RabbitmqApplication { static final String topicExchangeName = "spring-boot-exchange"; static final String queueName = "spring-boot"; @Bean Queue queue(){ return new Queue(queueName, false); } @Bean TopicExchange exchange(){ return new TopicExchange(topicExchangeName); } @Bean Binding binding(Queue queue,TopicExchange exchange){ return BindingBuilder.bind(queue).to(exchange).with("foo.bar.#"); } @Bean SimpleMessageListenerContainer container(ConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter){ SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(); container.setConnectionFactory(connectionFactory); container.setQueueNames(queueName); container.setMessageListener(listenerAdapter); return container; } @Bean MessageListenerAdapter listenerAdapter(Receiver receiver){ return new MessageListenerAdapter(receiver, "receiveMessage"); } public static void main(String[] args) { SpringApplication.run(RabbitmqApplication.class, args).close(); } }
@SpringBootApplication 是一個(gè)便利注釋,添加了以下所有內(nèi)容:
@Configuration 標(biāo)記該類作為應(yīng)用程序上下文的bean定義的源。
@EnableAutoConfiguration 告訴Spring Boot開始根據(jù)類路徑設(shè)置,其他bean和各種屬性設(shè)置添加bean。
通常你會(huì)添加@EnableWebMvc一個(gè)Spring MVC應(yīng)用程序,但Spring Boot會(huì)在類路徑上看到spring-webmvc時(shí)自動(dòng)添加它。這會(huì)將應(yīng)用程序標(biāo)記為Web應(yīng)用程序并激活關(guān)鍵行為,例如設(shè)置a DispatcherServlet。
@ComponentScan告訴Spring在包中尋找其他組件,配置和服務(wù)hello,允許它找到控制器。
該main()方法使用Spring Boot的SpringApplication.run()方法來(lái)啟動(dòng)應(yīng)用程序。您是否注意到?jīng)]有一行XML?也沒有web.xml文件。此Web應(yīng)用程序是100%純Java,您無(wú)需處理配置任何管道或基礎(chǔ)結(jié)構(gòu)。
listenerAdapter()方法中定義的bean在定義的容器中注冊(cè)為消息偵聽器container()。它將偵聽“spring-boot”隊(duì)列中的消息。因?yàn)镽eceiver該類是POJO,所以需要將其包裝在MessageListenerAdapter指定要調(diào)用的位置receiveMessage。
JMS隊(duì)列和AMQP隊(duì)列具有不同的語(yǔ)義。例如,JMS僅向一個(gè)使用者發(fā)送排隊(duì)的消息。雖然AMQP隊(duì)列執(zhí)行相同的操作,但AMQP生成器不會(huì)直接向隊(duì)列發(fā)送消息。相反,消息被發(fā)送到交換機(jī),交換機(jī)可以轉(zhuǎn)到單個(gè)隊(duì)列,或扇出到多個(gè)隊(duì)列,模仿JMS主題的概念。
消息監(jiān)聽器容器和接收器bean是您監(jiān)聽消息所需的全部?jī)?nèi)容。要發(fā)送消息,您還需要一個(gè)Rabbit模板。
該queue()方法創(chuàng)建AMQP隊(duì)列。該exchange()方法創(chuàng)建主題交換。該binding()方法將這兩者綁定在一起,定義RabbitTemplate發(fā)布到交換時(shí)發(fā)生的行為。
Spring AMQP要求將the Queue,the TopicExchange,和Binding聲明為頂級(jí)Spring bean才能正確設(shè)置。
在這種情況下,我們使用主題交換,并且隊(duì)列與路由密鑰綁定,foo.bar.#這意味著使用以路由鍵開頭的任何消息foo.bar.將被路由到隊(duì)列。
發(fā)送測(cè)試消息
測(cè)試消息由CommandLineRunner,他還等待接收器中的鎖存器并關(guān)閉應(yīng)用程序上下文:
@Slf4j @Component public class Runner implements CommandLineRunner { private final RabbitTemplate rabbitTemplate; private final Receiver receiver; public Runner(Receiver receiver, RabbitTemplate rabbitTemplate){ this.receiver = receiver; this.rabbitTemplate = rabbitTemplate; } @Override public void run(String... strings) throws Exception { log.info("Sending message...."); rabbitTemplate.convertAndSend(RabbitmqApplication.topicExchangeName,"foo.bar.baz","Hello from RabbitMQ!"); receiver.getLatch().await(10000, TimeUnit.MILLISECONDS); } }
請(qǐng)注意,模板將消息路由到交換機(jī),其路由密鑰foo.bar.baz與綁定匹配。
可以在測(cè)試中模擬出運(yùn)行器,以便可以單獨(dú)測(cè)試接收器。
運(yùn)行程序,你應(yīng)該看到如下輸出:
2018-12-03 10:23:46.779 INFO 10828 --- [ main] o.s.b.w.embedded.tomcat.TomcatWebServer : Tomcat started on port(s): 8080 (http) with context path '' 2018-12-03 10:23:46.782 INFO 10828 --- [ main] c.g.unclecatmyself.RabbitmqApplication : Started RabbitmqApplication in 3.61 seconds (JVM running for 4.288) 2018-12-03 10:23:46.784 INFO 10828 --- [ main] com.github.unclecatmyself.Runner : Sending message.... 2018-12-03 10:23:46.793 INFO 10828 --- [ container-1] com.github.unclecatmyself.Receiver : Received < Hello from RabbitMQ! > 2018-12-03 10:23:46.799 INFO 10828 --- [ main] o.s.a.r.l.SimpleMessageListenerContainer : Waiting for workers to finish. 2018-12-03 10:23:47.813 INFO 10828 --- [ main] o.s.a.r.l.SimpleMessageListenerContainer : Successfully waited for workers to finish. 2018-12-03 10:23:47.815 INFO 10828 --- [ main] o.s.s.concurrent.ThreadPoolTaskExecutor : Shutting down ExecutorService 'applicationTaskExecutor' 2018-12-03 10:23:47.816 INFO 10828 --- [ main] o.s.a.r.l.SimpleMessageListenerContainer : Shutdown ignored - container is not active already
關(guān)于怎么在spring中使用RabbitMQ傳遞消息問(wèn)題的解答就分享到這里了,希望以上內(nèi)容可以對(duì)大家有一定的幫助,如果你還有很多疑惑沒有解開,可以關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道了解更多相關(guān)知識(shí)。