SpringBoot集成ActiveMQ實體詳解-有解無憂

在專案開發的程序中我們經常會遇到類似的業務場景：用戶申請提現，后臺進行賬務處理、發送提現短信、呼叫銀行打款通道，

在這個程序中呼叫三方通道（短信或銀行通道）都比較耗時，同時賬務處理可能也是由專門的賬務系統進行處理，那么，為了提高并發和相應速度，后面的三個操作都可以通過異步進行處理，這就用到了訊息佇列，

訊息佇列中間件是分布式系統中重要的組件，主要解決應用耦合、異步訊息、流量削鋒等問題，實作高性能、高可用、可伸縮和最終一致性架構，是大型分布式系統不可缺少的中間件，

市面上比較常見的訊息佇列有：ActiveMQ、RabbitMQ、ZeroMQ、Kafka、MetaMQ、RocketMQ，

在Spring Boot的starter中專門集成了ActiveMQ，因此，本篇文章我們就來講講對ActiveMQ的集成，

JMS規范

JMS即Java訊息服務（Java Message Service）應用程式介面，是一個Java平臺中關于面向訊息中間件（MOM）的API，用于在兩個應用程式之間，或分布式系統中發送訊息，進行異步通信，Java訊息服務是一個與具體平臺無關的API，絕大多數MOM提供商都對JMS提供支持，

JMS的訊息機制有2種模型，一種是佇列的形式（Point to Point—）發送的訊息只能被一個消費者消費；一種是訂閱（Topic）模式，可以被多個訂閱者訂閱，訂閱者都會接收到同樣的訊息，

而ActiveMQ就是對JMS的實作之一，

ActiveMQ介紹

ActiveMQ是一種開源的基于JMS（Java Message Servie）規范的一種訊息中間件的實作，ActiveMQ的設計目標是提供標準的、面向訊息的、能夠跨越多語言和多系統的應用集成訊息通信中間件，

它為企業應用中訊息傳遞提供高可用、出色性能、可擴展、穩定和安全保障，

ActiveMQ實作JMS規范并在此之上提供大量額外的特性，ActiveMQ支持佇列和訂閱兩種模式的訊息發送，

AcitveMQ的資料傳送流程如下圖：

ActiveMQ的兩種訊息傳遞型別：

（1）點對點傳輸，即一個生產者對應一個消費者，生產者向broke推送資料，資料存盤在broke的一個佇列中，當消費者接受該條佇列里的資料，

（2）基于發布/訂閱模式的傳輸，即根據訂閱話題來接收相應資料，一個生產者可向多個消費者推送資料，與MQTT協議的實作是類似的，

兩種訊息傳遞型別的不同，點對點傳輸消費者可以接收到在連接之前生產者所推送的資料，而基于發布/訂閱模式的傳輸方式消費者只能接收到連接之后生產者推送的資料，

Spring Boot集成ActiveMQ

Spring Boot針對ActiveMQ專門提供了spring-boot-starter-activemq，用來支持ActiveMQ在Spring Boot的自動集成配置，在此基礎上我們可以很輕易的進行集成和使用，

創建專案并引入依賴

創建標準的Spring Boot專案，并在專案中引入以下依賴：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-activemq</artifactId>
</dependency>

此時如果不需要web或其他相關處理，只引入該依賴即可，如果使用pool的話, 就需要在pom中加入以下依賴：

<dependency>
     <groupId>org.apache.activemq</groupId>
     <artifactId>activemq-pool</artifactId>
</dependency>

組態檔

在application.properties中添加如下配置：

# 基于記憶體的ActiveMQ
spring.activemq.in-memory=true
# 不使用連接池，如果使用連接池還需在pom中添加activemq-pool的依賴
spring.activemq.pool.enabled=false

# 獨立安裝的ActiveMQ
#spring.activemq.broker-url=tcp://127.0.0.1:61616
#spring.activemq.user=admin
#spring.activemq.password=admin

上述配置中有兩套配置，Spring Boot支持基于記憶體ActiveMQ和基于獨立安裝的ActiveMQ，正常請求基于記憶體的形式是為了方便測驗而使用，基于獨立安裝的形式才是真正用于生產環境，此處為了講解功能，方便測驗，采用基于記憶體的形式，

佇列模式實體

首先，我們來實作基于佇列（Queue）形式的實作，這里需要用到兩個類ActiveMQQueue和JmsMessagingTemplate，前者是由ActiveMQ對javax.jms.Queue的介面實作，后者為Spring提供發送訊息的工具類，結合Queue對訊息進行發送，

JmsMessagingTemplate默認已經被實體化，直接拿來使用即可，而ActiveMQQueue則需要我們進行實體化，并傳入訊息佇列的名稱，

@Configuration
public class MyMqConfig {

	@Bean
	public Queue queue() {
		return new ActiveMQQueue("sms.queue");
	}
}

Spring Boot中很常規的實體化操作，不再贅述，當實體化完ActiveMQQueue之后，我們的佇列便創建完成，下面創建對應的生產者和消費者，

生產者對應代碼如下：

@Component
public class Producer {

	@Resource
	private JmsMessagingTemplate jmsMessagingTemplate;

	@Resource
	private Queue queue;

	public void sendMsg(String msg) {
		System.out.println("發送訊息內容 :" + msg);
		this.jmsMessagingTemplate.convertAndSend(this.queue, msg);
	}

}

此處用到JmsMessagingTemplate和Queue，上面已經提到，這兩個類都已經完成了初始化，消費者對應的配置如下：

@Component
public class Consumer {

	@JmsListener(destination = "sms.queue")
	public void receiveMsg(String text) {
		System.out.println("接收到訊息 : "+text);
	}
}

Spring提供了注解式監聽器端點：使用@JmsListener，使用@JmsListener托管bean的帶注釋方法對其進行訂閱，在Java8中，@JmsListener是一個可重復的注解，可以關聯多個JMS destinations到同一個方法中，而在Java 6和7中，可以使用@JmsListeners注解，

其中destination指定監控的訊息佇列名稱為“sms.queue”，當佇列sms.queue中有訊息發送時會觸發此方法的執行，text為訊息內容，

上面完成了佇列初始化、生產者和消費者代碼的撰寫，下面通過單元測驗來驗證是否能夠正確發送和處理訊息，

@RunWith(SpringRunner.class)
@SpringBootTest
public class ActiveMqTests {

	@Autowired
	private Producer producer;

	@Test
	public void sendSimpleQueueMessage() {
		this.producer.sendMsg("提現200.00元");
	}
}

執行單元測驗，會發現在日志中列印如下資訊：

發送訊息內容 :提現200.00元
接收到訊息 : 提現200.00元

說明訊息可以正常發送和接收，如果是基于記憶體模式，在執行單元測驗時會列印出“javax.jms.JMSException: peer (vm://localhost#1) stopped.”例外日志，這是Info級別的錯誤，是ActiveMQ的一個bug，

訂閱模式實體

廣播發送的訊息，可以被多個消費者接收，這里我們就在原有的基礎上進行廣播訊息的添加，

首先，Spring Boot集成ActiveMQ時默認只支持佇列或者廣播之一，通過配置項spring.jms.pub-sub-domain來指定，true 為廣播模式，false為佇列模式，默認情況下支持佇列模式，

此時要使用廣播模式，則需在組態檔中添加如下配置：

spring.jms.pub-sub-domain=true

需要注意的是，此時佇列模式不可正常作業，

然后在MyMqConfig中添加：

@Bean
public Topic topic() {
	return new ActiveMQTopic("sms.topic");
}

這里創建了ActiveMQTopic，并將topic的名稱指定為sms.topic，

Producer中新增如下代碼：

@Resource
private Topic topic;

public void sendTopic(String msg) {
	System.out.println("發送Topic訊息內容 :"+msg);
	this.jmsMessagingTemplate.convertAndSend(this.topic, msg);
}

為了演示多個廣播接收者，在Comsumer中新增兩個消費者：

@JmsListener(destination = "sms.topic")
public void receiveTopic1(String text) {
	System.out.println("receiveTopic1接收到Topic訊息 : " + text);
}

@JmsListener(destination = "sms.topic")
public void receiveTopic2(String text) {
	System.out.println("receiveTopic2接收到Topic訊息 : " + text);
}

單元測驗類中新增如下測驗：

@Test
public void sendSimpleTopicMessage() {
	this.producer.sendTopic("提現200.00元");
}

此時，執行單元測驗，便可看到如下日志資訊：

發送Topic訊息內容 :提現200.00元
receiveTopic2接收到Topic訊息 : 提現200.00元
receiveTopic1接收到Topic訊息 : 提現200.00元

說明訊息發送成功，

同時支持兩種形式

在上面的實體中，要么支持佇列模式要么支持廣播模式，如果在生產環境中兩者都需要支持，那么就需要自定義JmsListenerContainerFactory實體，當然，如果Spring Boot默認的配置無法滿足需求，也可以自定義該類，這里只是其中場景之一，

基本配置和使用步驟：通過DefaultJmsListenerContainerFactory創建自定義的JmsListenerContainerFactory實體，在@JmsListener注解中通過containerFactory屬性進行參考，

在MyMqConfig配置類中新增如下配置：

@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;
}

這里分別實體化了基于佇列和訂閱的工廠類，然后分別在對應的消費者方法上添加containerFactory屬性，示例代碼如下：

@JmsListener(destination = "sms.queue", containerFactory = "queueListenerFactory")
public void receiveMsg(String text) {
	System.out.println("接收到訊息 : " + text);
}

@JmsListener(destination = "sms.topic", containerFactory = "topicListenerFactory")
public void receiveTopic1(String text) {
	System.out.println("receiveTopic1接收到Topic訊息 : " + text);
}

分別執行兩種形式的訊息，發現都正常互利，同時，此時組態檔中的項spring.jms.pub-sub-domain也無效了，

其他事項

1、activeMq的埠號是61616；

2、使用topic,需要配置spring.jms.pub-sub-domain=true;

3、queue如果沒有消費者，會將資訊存盤到queue中；

4、發送的訊息為物件的時候，需要將物件序列化；消費者接收物件資訊時需要使用ObjectMessage進行轉化；

5、使用JmsListener注解中的containerFactory屬性，可以配置spring.jms.pub-sub屬性，實作同時接收queque和topic；

6、queue為點對點模式；tipic為發布訂閱模式；

7、示例中的訊息佇列名稱（sms.queue和sms.topic）可根據需要設定成配置屬性；

原始碼地址：https://github.com/secbr/springboot-learn/tree/master/springboot-activemq

參考文章：

https://www.cnblogs.com/xiguadadage/p/11217604.html

https://blog.csdn.net/bihansheng2010/article/details/87190645

程式新視界：精彩和成長都不容錯過

程式新視界-微信公眾號

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/69351.html

標籤：Java

上一篇：compare

下一篇：Java并發--final關鍵字