整合springboot · JAVA

[TOC] # JMS JMS 即 Java 消息服務（Java Message Service）應?程序接?，是?個 Java 平臺中關于?向消息中間件（MOM）的 API，?于在兩個應?程序之間，或分布式系統中發送消息，進?異步通信。Java 消息服務是 ?個與具體平臺?關的 API，絕?多數 MOM 提供商都對 JMS 提供?持 JMS 的消息機制有 2 種模型，?種是 Point to Point，表現為隊列的形式，發送的消息，只能被?個接收者取 ?；另?種是 Topic，可以被多個訂閱者訂閱，類似于群發 ActiveMQ 是 JMS 的?個實現 # 依賴 ~~~ <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-activemq</artifactId> </dependency> ~~~ # 配置文件使? ActiveMQ 時有兩種使??式，?種是使?獨?安裝的 ActiveMQ，在?產環境推薦使?這種；另?種是使?基于內存 ActiveMQ ，在調試階段建議使?這種?式。 ~~~ # 基于內存的 ActiveMQ spring.activemq.in-memory=true # 不適應連接池 spring.activemq.pool.enabled=false #獨?安裝的 ActiveMQ #spring.activemq.broker-url=tcp://192.168.0.1:61616 #spring.activemq.user=admin #spring.activemq.password=admin #spring.activemq.pool.enabled=true #spring.activemq.pool.max-connections=50 #spring.activemq.packages.trust-all=true ~~~ # 同時支持隊列和廣播 Spring Boot 集成 ActiveMQ 的項?默認只?持隊列或者?播中的?種，通過配置項 `spring.jms.pub-sub-domain` 的值來控制，true 為?播模式，false 為隊列模式，默認情況下?持隊列模式。如果需要在同?項?中既?持隊列模式也?持?播模式，可以通過DefaultJmsListenerContainerFactory創建?定義的 JmsListenerContainerFactory 實例，之后在 `@JmsListener`注解中通過 containerFactory屬性引 ?它分別創建兩個?定義的 JmsListenerContainerFactory 實例，通過 pubSubDomain 來控制是?持隊列模式還是?播模式 **然后在消費者接收的?法中，指明使? containerFactory 接收消息** ~~~ package com.jdxia.config; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.jms.annotation.EnableJms; import org.springframework.jms.config.DefaultJmsListenerContainerFactory; import org.springframework.jms.config.JmsListenerContainerFactory; import javax.jms.ConnectionFactory; @Configuration @EnableJms public class ActiveMQConfig { @Bean("queueListenerFactory") public JmsListenerContainerFactory<?> queueListenerFactory(ConnectionFactory connectionFactory) { DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory(); factory.setConnectionFactory(connectionFactory); factory.setPubSubDomain(false); return factory; } @Bean("topicListenerFactory") public JmsListenerContainerFactory<?> topicListenerFactory(ConnectionFactory connectionFactory) { DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory(); factory.setConnectionFactory(connectionFactory); factory.setPubSubDomain(true); return factory; } } ~~~ # 隊列 ## 創建隊列定義了?個隊列 queue 命名為：neo.queue ~~~ package com.jdxia.config; import org.apache.activemq.command.ActiveMQQueue; import org.apache.activemq.command.ActiveMQTopic; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import javax.jms.Queue; import javax.jms.Topic @Configuration // 在項?啟動時定義.... public class MqConfig { @Bean public Queue queue() { //隊列 return new ActiveMQQueue("neo.queue"); } } ~~~ ## 創建生產者 ~~~ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.jms.core.JmsMessagingTemplate; import org.springframework.stereotype.Component; import javax.jms.Queue; import javax.jms.Topic; //生產者 @Component public class Producer { //Spring 提供發送消息的?具類 @Autowired private JmsMessagingTemplate jmsMessagingTemplate; @Autowired private Queue queue; @Autowired private Topic topic; public void sendQueue(String msg) { System.out.println("send queue msg :" + msg); //轉換并發送 this.jmsMessagingTemplate.convertAndSend(this.queue, msg); } } ~~~ ## 創建消費者(多) 當有多個消費者監聽?個隊列時，消費者會?動均衡負載的接收消息，并且每個消息只能有?個消費者所接收。 ~~~ package com.jdxia.consumer; import org.springframework.jms.annotation.JmsListener; import org.springframework.stereotype.Component; @Component public class Consumer { // 監控了名為 neo.queue 的隊 @JmsListener(destination = "neo.queue", containerFactory = "queueListenerFactory") public void receiveQueue(String text) { System.out.println("Consumer queue msg : " + text); } } ~~~ ~~~ package com.jdxia.consumer; import org.springframework.jms.annotation.JmsListener; import org.springframework.stereotype.Component; @Component public class Consumer2 { @JmsListener(destination = "neo.queue", containerFactory = "queueListenerFactory") public void receiveQueue(String text) { System.out.println("Consumer2 msg : "+text); } } ~~~ ## 測試 ~~~ @Autowired private Producer producer; //OutputCapture 是 Spring Boot 提供的?個測試類，它能捕獲 System.out 和 System.err 的輸出，我們可以利 ?這個特性來判斷程序中的輸出是否執? @Rule public OutputCapture outputCapture = new OutputCapture(); @Test public void test() { for (int i = 0; i < 100; i++) { this.producer.sendQueue("Test queue message" + i); } Thread.sleep(1000L); System.out.println(this.outputCapture.toString().contains("Test queue")); } ~~~ # 廣播 ?播（Topic）是?個發送者多個消費者的模式 ## 創建廣播定義了?個?播 Topic 命名為：neo.topic ~~~ package com.jdxia.config; import org.apache.activemq.command.ActiveMQQueue; import org.apache.activemq.command.ActiveMQTopic; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import javax.jms.Queue; import javax.jms.Topic; @Configuration // 在項?啟動時定義.... public class MqConfig { @Bean public Topic topic() { return new ActiveMQTopic("neo.topic"); } } ~~~ ## 創建生產者 ~~~ package com.jdxia.producer; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.jms.core.JmsMessagingTemplate; import org.springframework.stereotype.Component; import javax.jms.Queue; import javax.jms.Topic; //生產者 @Component public class Producer { //Spring 提供發送消息的?具類 @Autowired private JmsMessagingTemplate jmsMessagingTemplate; @Autowired private Queue queue; @Autowired private Topic topic; public void sendTopic(String msg) { System.out.println("send topic msg :" + msg); this.jmsMessagingTemplate.convertAndSend(this.topic, msg); } } ~~~ ## 創建消費者 ~~~ package com.jdxia.consumer; import org.springframework.jms.annotation.JmsListener; import org.springframework.stereotype.Component; @Component public class Consumer { @JmsListener(destination = "neo.topic", containerFactory = "topicListenerFactory") public void receiveTopic(String text) { System.out.println("Consumer topic msg : " + text); } } ~~~ ## 測試 ~~~ @Autowired private Producer producer; //OutputCapture 是 Spring Boot 提供的?個測試類，它能捕獲 System.out 和 System.err 的輸出，我們可以利 ?這個特性來判斷程序中的輸出是否執? @Rule public OutputCapture outputCapture = new OutputCapture(); @Test public void test() throws InterruptedException { this.producer.sendTopic("Test Topic message"); Thread.sleep(1000L); } ~~~ ~~~ send topic msg :Test Topic message Consumer topic msg : Test Topic message Consumer2 topic msg : Test Topic message ~~~