04.ActiveMQ與Spring JMS整合

    SpringJMS使用參考：http://docs.spring.io/spring/docs/current/spring-framework-reference/html/jms.html 

1.SpringJMS核心接口介紹

1.JmsTemplate

    JmsTemplate: 是Spring自身提供，只需向Spring容器內注冊這個類即可，就可以使用 JmsTemplate類對象方便的操作JMS，下面介紹他常用的方法。

    注意： JmsTemplate類是線程安全的，可以在整個應用范圍使用。但並不是說整個引用只能使用一個JmsTemplate實例，可以根據需要注入多個JmsTemplate實例。

1. // send - 發送一個消息，使用消息創建接口MessageCreator
2. publicvoid send(MessageCreator messageCreator)
3. publicvoid send(finalDestination destination,finalMessageCreator messageCreator)
4. publicvoid send(finalString destinationName,finalMessageCreator messageCreator)
6. // sendAndReceive - 發送並接收消息
7. publicMessage sendAndReceive(MessageCreator messageCreator)
8. publicMessage sendAndReceive(finalDestination destination,finalMessageCreator messageCreator)
9. publicMessage sendAndReceive(finalString destinationName,finalMessageCreator messageCreator)
11. // convertAndSend - 使用MessageConverter接口轉換消息，先將對象轉換成消息再發送消息。與之對應的是receiveAndConvert方法
12. publicvoid convertAndSend(Object message)throwsJmsException
13. publicvoid convertAndSend(Destination destination,finalObject message)
14. publicvoid convertAndSend(String destinationName,finalObject message)
15. publicvoid convertAndSend(Object message,MessagePostProcessor postProcessor)
16. publicvoid convertAndSend(Destination destination,finalObject message,finalMessagePostProcessor postProcessor)
17. publicvoid convertAndSend(String destinationName,finalObject message,finalMessagePostProcessor postProcessor)
19. // receive - 接收消息
20. publicMessage receive()
21. publicMessage receive(Destination destination)
22. publicMessage receive(String destinationName)
24. // receiveSelected - 接收消息，並過濾消息
25. publicMessage receiveSelected(String messageSelector)
26. publicMessage receiveSelected(finalDestination destination,finalString messageSelector)
27. publicMessage receiveSelected(finalString destinationName,finalString messageSelector)
29. // receiveAndConvert - 接收消息，並使用MessageConverter接口轉換消息，把一個消息轉換成一個對象
30. publicObject receiveAndConvert()
31. publicObject receiveAndConvert(Destination destination)
32. publicObject receiveAndConvert(String destinationName)
34. // receiveSelectedAndConvert - 接收消息，並使用消息過濾和消息轉換
35. publicObject receiveSelectedAndConvert(String messageSelector)
36. publicObject receiveSelectedAndConvert(Destination destination,String messageSelector)
37. publicObject receiveSelectedAndConvert(String destinationName,String messageSelector)
39. // browse - 瀏覽消息
40. public<T> T browse(BrowserCallback<T> action)
41. public<T> T browse(Queue queue,BrowserCallback<T> action)
42. public<T> T browse(String queueName,BrowserCallback<T> action)
44. // browseSelected - 瀏覽消息，並使用過濾
45. public<T> T browseSelected(String messageSelector,BrowserCallback<T> action)
46. public<T> T browseSelected(finalQueue queue,finalString messageSelector,finalBrowserCallback<T> action)
47. public<T> T browseSelected(finalString queueName,finalString messageSelector,finalBrowserCallback<T> action)
49. // execute - 執行SessionCallback、ProducerCallback、BrowserCallback回調接口，並得到回調接口返回值
50. public<T> T execute(SessionCallback<T> action)
51. public<T> T execute(SessionCallback<T> action,boolean startConnection)
52. public<T> T execute(ProducerCallback<T> action)
53. public<T> T execute(finalDestination destination,finalProducerCallback<T> action)
54. public<T> T execute(finalDestination destination,finalProducerCallback<T> action)

2.連接工廠(連接池)

    JmsTemplate需要引用一個ConnectionFactory， JmsTemlate每次發送消息時都會重新創建連接，創建connection，session，創建productor。這是一個非常耗性能的地方，特別是大數據量的情況下。所以出現了 PooledConnectionFactory， PooledConnectionFactory是 ActiveMQ中的類，它實現了ConnectionFactory。 而 SpringJMS的實現有 SingleConnectionFactory、 CachingConnectionFactory。

1. PooledConnectionFactory：會緩存connection，session和productor，不會緩存consumer。因此只適合於生產者發送消息。
2. SingleConnectionFactory：對於建立JMS服務器鏈接的請求會一直返回同一個Connection，並且會忽略Connection的close方法調用(包括調用createConnection()得到的Connection)
3. CachingConnectionFactory：繼承了SingleConnectionFactory，所以它擁有SingleConnectionFactory的所有功能，同時它還新增了緩存功能，它可以緩存Session、MessageProducer和MessageConsumer。

    SpringJMS提供了CachingConnectionFactory，這才是首選的方案。然而CachingConnectionFactory有一個問題必須指出，默認情況下，CachingConnectionFactory只緩存一個session，在它的JavaDoc中，它聲明對於低並發情況下這是足夠的。與之相反，PooledConnectionFactory的默認值是500。這些設置，在很多情況下，需要親自去測試並驗證。我將其設置為100，對我來說還是很不錯。

3.接口介紹

以下是SpringJMS的常用接口或類：

1. MessageCreator -- 消息創建接口，發送消息時需要使用此接口創建消息
2. SessionCallback -- 使用JMS Session時的回調接口
3. ProducerCallback -- 使用JMS Session和MessageProducer時的回調接口
4. BrowserCallback -- 使用JMS Session和QueueBrowser時的回調接口
5. MessageListener -- 消息監聽器接口，注解@JmsListener與其功能相似
6. ListenerContainer -- 消息偵聽器容器接口，實現有SimpleMessageListenerContainer、DefaultMessageListenerContainer
7. MessageConverter -- 消息轉換接口，用於JMS消息到Java對象之間的轉換

2.消息監聽器

    在Spring整合JMS的應用中我們在定義消息監聽器的時候一共可以定義三種類型的消息監聽器，分別是MessageListener、SessionAwareMessageListener和MessageListenerAdapter。下面就分別來介紹一下這幾種類型的區別。

1.MessageListener

    MessageListener是最原始的消息監聽器，它是JMS規范中定義的一個接口。其中定義了一個用於處理接收到的消息的onMessage方法，該方法只接收一個Message參數。我們前面在講配置消費者的時候用的消息監聽器就是MessageListener，代碼如下：

1. publicclassQueueMessageListenerimplementsMessageListener
2. {
3. @Override
4. publicvoid onMessage(Message message)
5. {
6. try
7. {
8. if(message instanceofTextMessage)
9. {
10. TextMessage tm =(TextMessage) message;
11. System.out.println("監聽消息："+ tm.getText());
12. }
13. else
14. {
15. System.out.println("消息類型："+ message.getClass());
16. }
17. }
18. catch(JMSException e)
19. {
20. e.printStackTrace();
21. }
22. }
23. }

     在上面代碼中我們只是簡單的打印了一些消息相關的信息。

2.SessionAwareMessageListener

    SessionAwareMessageListener是Spring為我們提供的，它不是標准的JMS MessageListener。MessageListener的設計只是純粹用來接收消息的，假如我們在使用MessageListener處理接收到的消息時我們需要發送一個消息通知對方我們已經收到這個消息了，那么這個時候我們就需要在代碼里面去重新獲取一個Connection或Session。SessionAwareMessageListener的設計就是為了方便我們在接收到消息后發送一個回復的消息，它同樣為我們提供了一個處理接收到的消息的onMessage方法，但是這個方法可以同時接收兩個參數，一個是表示當前接收到的消息Message，另一個就是可以用來發送消息的Session對象。先來看一段代碼：

1. publicclassQueueSessionAwareMessageListenerimplementsSessionAwareMessageListener<TextMessage>
2. {
3. /** 回復消息的目的地 */
4. privateDestination destination;
6. @Override
7. publicvoid onMessage(TextMessage message,Session session)throwsJMSException
8. {
9. System.out.println("監聽消息內容："+ message.getText());
10. MessageProducer messageProducer = session.createProducer(destination);
11. TextMessage replyMessage = session.createTextMessage("已收到消息："+ message.getJMSMessageID());
12. messageProducer.send(replyMessage);
13. }
15. publicDestination getDestination()
16. {
17. return destination;
18. }
20. publicvoid setDestination(Destination destination)
21. {
22. this.destination = destination;
23. }
24. }

    在上面代碼中我們定義了一個SessionAwareMessageListener，在這個Listener中我們在接收到了一個消息之后，利用對應的Session創建了一個到destination的生產者和對應的消息，然后利用創建好的生產者發送對應的消息。

3.MessageListenerAdapter

    MessageListenerAdapter類實現了MessageListener接口和SessionAwareMessageListener接口，它的主要作用是將接收到的消息進行類型轉換，然后通過反射的形式把它交給一個普通的Java類進行處理。 MessageListenerAdapter會把接收到的消息做如下轉換：

1. TextMessage轉換為String對象
2. BytesMessage轉換為byte數組
3. MapMessage轉換為Map對象
4. ObjectMessage轉換為對應的Serializable對象

    既然前面說了MessageListenerAdapter會把接收到的消息做一個類型轉換，然后利用反射把它交給真正的目標處理器——一個普通的Java類進行處理，如果真正的目標處理器是一個MessageListener或者是一個SessionAwareMessageListener，那么Spring將直接使用接收到的Message對象作為參數調用它們的onMessage方法，而不會再利用反射去進行調用。那么我們在定義一個MessageListenerAdapter的時候就需要為它指定這樣一個目標類。這個目標類我們可以通過MessageListenerAdapter的構造方法參數指定，也可以通過屬性 delegate 指定，如：

1. <!--消息監聽適配器-->
2. <bean id="messageListenerAdapter"class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
3. <!--通過構造函數指定消息處理器-->
4. <constructor-arg>
5. <bean id="readerMessage"class="com.test.ReaderMessage"/>
6. </constructor-arg>
7. <!--指定消息處理器類處理消息的方法-->
8. <property name="defaultListenerMethod" value="receiveMessage"/>
9. <!--通過delegate屬性指定消息處理器-->
10. <property name="delegate">
11. <bean id="readerMessage"class="com.test.ReaderMessage"/>
12. </property>
13. </bean>

    前面說了如果我們指定的這個目標處理器是一個MessageListener或者是一個SessionAwareMessageListener的時候Spring將直接利用接收到的Message對象作為方法參數調用它們的onMessage方法。但是如果指定的目標處理器是一個普通的Java類時Spring將利用Message進行了類型轉換之后的對象作為參數通過反射去調用真正的目標處理器的處理方法，那么Spring是如何知道該調用哪個方法呢？這是通過MessageListenerAdapter的defaultListenerMethod屬性來決定的，當我們沒有指定該屬性時，Spring會默認調用目標處理器的handleMessage方法。 若使用以下 MessageListenerByAdapter 類處理消息，可以配置 MessageListenerAdapter 的 delegate 屬性來設置處理消息的方法。

1. publicclassMessageListenerByAdapter
2. {
3. publicvoid handleMessage(String message)
4. {
5. System.out.println("handleMessage方法處理消息，消息內容是："+ message);
6. }
8. publicvoid receiveMessage(String message)
9. {
10. System.out.println("receiveMessage方法處理消息，消息內容是："+ message);
11. }
12. }

    MessageListenerAdapter除了會自動的把一個普通Java類當做MessageListener來處理接收到的消息之外， 其另外一個主要的功能是可以自動的發送返回消息。 當我們設置的消息處理方法的返回值不為空的時候，SpringJMS會自動將它封裝為一個JMS Message，然后自動進行回復。那么這個時候這個回復消息將發送到哪里呢？這主要有兩種方式可以指定。

1. 第一，可以通過發送的Message的setJMSReplyTo方法指定該消息對應的回復消息的目的地。這需要生產者發送消息之前調用setJMSReplyTo方法。
2. 第二，通過MessageListenerAdapter的defaultResponseDestination屬性來指定。

注意： 當兩種方式都指定了消息的回復目的地的時候使用發送消息的setJMSReplyTo方法指定的目的地將具有較高的優先級。

4.配置消息監聽器

    配置以上三種消息監聽器，需要使用消息偵聽容器(MessageListenerContainer )， 在配置一個MessageListenerContainer的時候有三個屬性必須指定，一個是表示從哪里監聽的ConnectionFactory；一個是表示監聽什么的Destination；一個是接收到消息以后進行消息處理的MessageListener。示例如下：

1. <bean id="jmsContainer"class="org.springframework.jms.listener.DefaultMessageListenerContainer">
2. <!--設置連接工廠-->
3. <property name="connectionFactory" ref="connectionFactory"/>
4. <!--設置監聽地址-->
5. <property name="destination" ref="queueDestination"/>
6. <!--設置消息監聽處理器，可以是以上三種-->
7. <property name="messageListener" ref="queueMessageListener"/>
8. </bean>

Spring提供了三種AbstractMessageListenerContainer的子類，每種各有其特點。

1. SimpleMessageListenerContainer：這個消息偵聽容器是三種中最簡單的。它在啟動時創建一個會話session和消費者Consumer，並且會使用標准的JMS MessageConsumer.setMessageListener()方法注冊監聽器讓JMS提供者調用監聽器的回調函數。它不會動態的適應運行時需要和參與外部的事務管理。兼容性方面，它非常接近於獨立的JMS規范，但一般不兼容Java EE的JMS限制。
   
2. DefaultMessageListenerContainer：這個消息偵聽器使用的最多。跟SimpleMessageListenerContainer相比，DefaultMessageListenerContainer會動態的適應運行時需要，並且能夠參與外部的事務管理。它很好的平衡了對JMS提供者要求低、先進功能如事務參與和兼容Java EE環境。
   
3. ServerSessionMessageListenerContainer：這個監聽器容器利用JMS ServerSessionPool SPI動態管理JMS Session。使用者各種消息監聽器可以獲得運行時動態調優功能，但是這也要求JMS提供者支持ServerSessionPool SPI。如果不需要運行時性能調整，請使用 DefaultMessageListenerContainer 或 SimpleMessageListenerContainer。

3.ActiveMQ與SpringJMS整合示例

    本節提供了一個SpringJMS與ActiveMQ 整合的示例 ，本示例可以作為參考。

1.配置spring-context-jms.xml

1. <?xml version="1.0" encoding="UTF-8"?>
2. <beansxmlns="http://www.springframework.org/schema/beans"
3. xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4. xmlns:context="http://www.springframework.org/schema/context"
5. xmlns:jms="http://www.springframework.org/schema/jms"
6. xsi:schemaLocation="
7. http://www.springframework.org/schema/jms
8. http://www.springframework.org/schema/jms/spring-jms-4.0.xsd
9. http://www.springframework.org/schema/beans
10. http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
11. http://www.springframework.org/schema/context
12. http://www.springframework.org/schema/context/spring-context-4.0.xsd"
13. default-lazy-init="false">
14. <!-- 配置JMS連接工廠 -->
15. <beanid="connectionFactory"class="org.apache.activemq.ActiveMQConnectionFactory">
16. <propertyname="brokerURL"value="failover:(tcp://localhost:61616)"/>
17. </bean>
18. <!-- ActiveMQ連接池配置,ActiveMQ實現 -->
19. <!--
20. <bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">
21. <property name="connectionFactory" ref="connectionFactory" />
22. </bean>
23. -->
24. <!-- ActiveMQ連接池配置,SpingJMS實現 -->
25. <beanid="cachingConnectionFactory"class="org.springframework.jms.connection.CachingConnectionFactory">
26. <propertyname="targetConnectionFactory"ref="connectionFactory"/>
27. <!-- Session緩存數量,這里屬性也可以直接在這里配置 -->
28. <propertyname="sessionCacheSize"value="100"/>
29. </bean>
30. <!-- 消息隊列01 -->
31. <beanid="queueDestination01"class="org.apache.activemq.command.ActiveMQQueue">
32. <!-- 設置消息隊列的名字 -->
33. <constructor-arg>
34. <value>spring-jms-queue01</value>
35. </constructor-arg>
36. </bean>
38. <!-- 消息隊列02 -->
39. <beanid="queueDestination02"class="org.apache.activemq.command.ActiveMQQueue">
40. <!-- 設置消息隊列的名字 -->
41. <constructor-arg>
42. <value>spring-jms-queue02</value>
43. </constructor-arg>
44. </bean>
45. <!-- 配置JMS模板，Spring提供的JMS工具類，它發送、接收消息。 -->
46. <beanid="jmsTemplate"class="org.springframework.jms.core.JmsTemplate">
47. <propertyname="connectionFactory"ref="cachingConnectionFactory"/>
48. <propertyname="defaultDestination"ref="queueDestination01"/>
49. <propertyname="receiveTimeout"value="10000"/>
50. </bean>
51. <!-- 消息監聽容器（Queue），配置連接工廠，監聽的隊列是spring-jms-queue02，監聽器是上面定義的監聽器 -->
52. <beanid="jmsContainer"class="org.springframework.jms.listener.DefaultMessageListenerContainer">
53. <propertyname="connectionFactory"ref="cachingConnectionFactory"/>
54. <propertyname="destination"ref="queueDestination02"/>
55. <propertyname="messageListener">
56. <beanid="queueMessageListener"class="spring.QueueMessageListener"/>
57. </property>
58. </bean>
59. </beans>

注意：在配置文件中使用了 SpringJMS的命名空間，就可以使用配置： <jms:listener-container>、 <jms:jca-listener-container>

2.代碼實現

配置文件中的消息監聽器 spring.QueueMessageListener 代碼如下：

1. publicclassQueueMessageListenerimplementsMessageListener
2. {
3. @Override
4. publicvoid onMessage(Message message)
5. {
6. try
7. {
8. if(message instanceofTextMessage)
9. {
10. TextMessage tm =(TextMessage) message;
11. System.out.println("監聽消息："+ tm.getText());
12. }
13. else
14. {
15. System.out.println("消息類型："+ message.getClass());
16. }
17. }
18. catch(JMSException e)
19. {
20. e.printStackTrace();
21. }
22. }
23. }

實現消息創建接口：

1. /** 自定義消息創建器 */
2. publicclassMyMessageCreatorimplementsMessageCreator
3. {
4. privateString messageStr;
5. publicMyMessageCreator(String messageStr)
6. {
7. this.messageStr = messageStr;
8. }
9. @Override
10. publicMessage createMessage(Session session)throwsJMSException
11. {
12. return session.createTextMessage(messageStr);
13. }
14. }

根據配置編寫測試代碼：

1. publicstaticvoid main(String[] args)throwsException
2. {
3. ClassPathXmlApplicationContext applicationContext =newClassPathXmlApplicationContext("spring-context-jms.xml");
4. // 得到消息隊列
5. Destination queueDestination01 = applicationContext.getBean("queueDestination01",Destination.class);
6. Destination queueDestination02 = applicationContext.getBean("queueDestination02",Destination.class);
7. // Spring JMS操作模版
8. JmsTemplate jmsTemplate = applicationContext.getBean("jmsTemplate",JmsTemplate.class);
10. // 發送消息
11. jmsTemplate.send(queueDestination01,newMyMessageCreator("測試消息001"));
12. jmsTemplate.send(queueDestination01,newMyMessageCreator("測試消息002"));
13. jmsTemplate.send(queueDestination02,newMyMessageCreator("測試消息003"));
14. jmsTemplate.send(queueDestination02,newMyMessageCreator("測試消息004"));
16. //接收消息
17. TextMessage textMessage =(TextMessage) jmsTemplate.receive(queueDestination01);
18. System.out.println(textMessage.getText());
19. textMessage =(TextMessage) jmsTemplate.receive(queueDestination01);
20. System.out.println(textMessage.getText());
22. applicationContext.close();
23. System.out.println("OK!");
24. }

    本示例在配置文件中配置了兩個隊列： queueDestination01 、 queueDestination02 ，並使用監聽器 queueMessageListener 監聽隊列 queueDestination02 。在測試代碼中分別向兩個隊列發送消息，然后接收隊列 queueDestination01 的消息，而隊列 queueDestination02 的消息會被監聽器 queueMessageListener 接收處理。

3.運行結果

1. 監聽消息：測試消息003
2. 監聽消息：測試消息004
3. 測試消息001
4. 測試消息002
5. OK!

4.消息轉換MessageConverter

    MessageConverter的作用主要有兩方面，一方面它可以把我們的非標准化Message對象轉換成我們的目標Message對象，這主要是用在發送消息的時候；另一方面它又可以把我們的Message對象轉換成對應的目標對象，這主要是用在接收消息的時候。

    例如：我們需要把一個用戶信息 User 對象當JMS消息一樣發送和接收，而不需要每次發送與接收之前都需要寫轉換代碼(序列化)，此時我們就可以使用MessageConverter 接口。如用戶 User 類定義如下：

1. publicclassUserimplementsSerializable
2. {
3. privatestaticfinallong serialVersionUID =1L;
4. privateString name;
5. privateint age;
6. privateboolean sex;
7. ......
8. }

1.實現轉換接口MessageConverter

    MessageConverter接口的實現如下，這里我們簡單的使用Json序列化和反序列化實現：

1. publicclassUserMessageConverterimplementsMessageConverter
2. {
3. privateObjectMapper mapper =newObjectMapper();
4. @Override
5. publicMessage toMessage(Object object,Session session)throwsJMSException,MessageConversionException
6. {
7. String json =null;
8. try
9. {
10. json = mapper.writeValueAsString(object);
11. }
12. catch(JsonProcessingException e)
13. {
14. e.printStackTrace();
15. }
16. System.out.println("toMessage : "+ json);
17. return session.createTextMessage(json);
18. }
19. @Override
20. publicObject fromMessage(Message message)throwsJMSException,MessageConversionException
21. {
22. String json =((TextMessage) message).getText();
23. Object object =null;
24. try
25. {
26. object = mapper.readValue(json,User.class);
27. }
28. catch(Exception e)
29. {
30. e.printStackTrace();
31. }
32. System.out.println("fromMessage : "+ object);
33. return object;
34. }
35. }

    Json的序列化與反序列化使用了 jackson庫。其實Spring也實現了幾種 MessageConverter，如下：

1. org.springframework.jms.support.converter.SimpleMessageConverter
2. org.springframework.jms.support.converter.MessagingMessageConverter
3. org.springframework.jms.support.converter.MarshallingMessageConverter
4. org.springframework.jms.support.converter.MappingJackson2MessageConverter

2.配置 MessageConverter

2. <!-- 配置JMS模板，Spring提供的JMS工具類，它發送、接收消息。 -->
3. <beanid="jmsTemplate"class="org.springframework.jms.core.JmsTemplate">
4. <propertyname="connectionFactory"ref="cachingConnectionFactory"/>
5. <propertyname="defaultDestination"ref="queueDestination01"/>
6. <propertyname="receiveTimeout"value="10000"/>
7. <propertyname="messageConverter">
8. <beanid="userMessageConverter"class="spring.UserMessageConverter"/>
9. </property>
10. </bean>

配置文件仍然基於之前的整合示例！

3.測試示例

    基於之前的整合示例編寫如下測試代碼：

1. User user=newUser("危常煥",20,true);
2. // 發送消息
3. jmsTemplate.convertAndSend(user);
4. //接收消息
5. Object object = jmsTemplate.receiveAndConvert();
6. System.out.println(object);

    注意：需要調用 convertAndSend 和 receiveAndConvert 方法發送和接收消息，運行結果如下：

1. toMessage :{"name":"危常煥","age":20,"sex":true}
2. fromMessage :User[name=危常煥, age=20, sex=true]
3. User[name=危常煥, age=20, sex=true]
4. OK!

5.事務管理JmsTransactionManager

 

 

 

 

 

 

 

 

-------------------------------------------------------------------------------------------------------------------------------

來自為知筆記(Wiz)