前言

本系列Demo均以SpringBoot快速構(gòu)建，基本包使用到lombok（一個(gè)便捷的對(duì)象構(gòu)造工具 get/set）、spring-boot-starter-web，使用SpringBoot僅為了快速構(gòu)建Sample項(xiàng)目，對(duì)于學(xué)習(xí)Spring的對(duì)應(yīng)功能無(wú)影響。

我們希望你已經(jīng)有一定的java基礎(chǔ)與了解一個(gè)自己喜歡的IDEA功能，謝謝。

GitHub

地址：https://github.com/UncleCatMySelf/Spring-Tutorial

學(xué)習(xí)

完成設(shè)置發(fā)布和訂閱消息的RabbitMQ AMQP服務(wù)器的過(guò)程。

構(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í)，您可以將其命名為任何您想要的名稱。

為方便起見(jiàn)，這個(gè)POJO也有一個(gè)CountDownLatch。這允許它發(fā)信號(hào)通知接收到消息。這是您不太可能在生產(chǎn)應(yīng)用程序中實(shí)現(xiàn)的。

注冊(cè)監(jiān)聽(tīng)器并發(fā)送消息

Spring AMQP RabbitTemplate 提供了使用RabbitMQ發(fā)送和接收消息所需的一切。具體來(lái)說(shuō)，你需要配置：

• 消息偵聽(tīng)器容器
• 聲明隊(duì)列，交換以及它們之間的綁定
• 用于發(fā)送一些消息以測(cè)試偵聽(tīng)器的組件

Spring Boot會(huì)自動(dòng)創(chuàng)建連接工廠和RabbitTemplate，從而減少您必須編寫(xiě)的代碼量。

您將使用RabbitTemplate發(fā)送消息，并將Receiver使用消息偵聽(tīng)器容器注冊(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開(kāi)始根據(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？也沒(méi)有web.xml文件。此Web應(yīng)用程序是100％純Java，您無(wú)需處理配置任何管道或基礎(chǔ)結(jié)構(gòu)。

listenerAdapter()方法中定義的bean在定義的容器中注冊(cè)為消息偵聽(tīng)器container()。它將偵聽(tīng)“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)聽(tīng)器容器和接收器bean是您監(jiān)聽(tīng)消息所需的全部?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.#這意味著使用以路由鍵開(kāi)頭的任何消息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

結(jié)尾

恭喜！您剛剛使用Spring和RabbitMQ開(kāi)發(fā)了一個(gè)簡(jiǎn)單的發(fā)布 - 訂閱應(yīng)用程序。

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

最新評(píng)論