JMS 之 Active MQ 的spring整合

来源:http://www.cnblogs.com/jalja/archive/2017/06/22/7067741.html
-Advertisement-
Play Games

ConnectionFactory是用於產生到JMS伺服器的鏈接的，Spring為我們提供了多個ConnectionFactory，有SingleConnectionFactory和CachingConnectionFactory。SingleConnectionFactory對於建立JMS伺服器鏈... ...


一、與spring整合實現ptp的同步接收消息

pom.xml:

   <!-- https://mvnrepository.com/artifact/org.springframework/spring-jms -->
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-jms</artifactId>
        <version>4.3.7.RELEASE</version>
    </dependency>
  <!-- https://mvnrepository.com/artifact/org.apache.activemq/activemq-pool -->
    <dependency>
        <groupId>org.apache.activemq</groupId>
        <artifactId>activemq-pool</artifactId>
        <version>5.9.0</version>
    </dependency>

spring-jms.xml:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:jms="http://www.springframework.org/schema/jms"
xmlns:amq="http://activemq.apache.org/schema/core"
xsi:schemaLocation="http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core.xsd
http://www.springframework.org/schema/jms
http://www.springframework.org/schema/jms/spring-jms.xsd
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd">

<!-- ActiveMQConnectionFactory就是JMS中負責創建到ActiveMQ連接的工廠類 -->
<bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory" >
     <property name="brokerURL" value="tcp://192.168.0.224:61616"/>  
</bean>
<!-- 創建連接池 -->
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">  
    <property name="connectionFactory" ref="connectionFactory"/>  
    <property name="maxConnections" value="10"/>  
</bean>  
<!-- Spring為我們提供了多個ConnectionFactory，有SingleConnectionFactory和CachingConnectionFactory -->
<bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">  
    <property name="targetConnectionFactory" ref="pooledConnectionFactory"/>  
</bean> 

<!-- Spring提供的JMS工具類，它可以進行消息發送、接收等 -->  
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">  
    <!-- 這個connectionFactory對應的是我們定義的Spring提供的那個ConnectionFactory對象 -->  
    <property name="connectionFactory" ref="cachingConnectionFactory"/>  
</bean> 

<!--這個是隊列目的地，點對點的-->  
<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">  
    <constructor-arg index="0" value="spring-queue"/> 
</bean>   

</beans>

   ConnectionFactory是用於產生到JMS伺服器的鏈接的，Spring為我們提供了多個ConnectionFactory，有SingleConnectionFactory和CachingConnectionFactory。SingleConnectionFactory對於建立JMS伺服器鏈接的請求會一直返回同一個鏈接，並且會忽略Connection的close方法調用。CachingConnectionFactory繼承了SingleConnectionFactory，所以它擁有SingleConnectionFactory的所有功能，同時它還新增了緩存功能，它可以緩存Session、MessageProducer和MessageConsumer。這裡我們使用CachingConnectionFactory來作為示例。

消息生產者：

package com.jalja.org.jms.spring;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;

public class SpringJmsSend {
    public static void main(String[] args) {
        ApplicationContext context=new ClassPathXmlApplicationContext("spring-jms.xml");
        JmsTemplate jmsTemplate=(JmsTemplate) context.getBean("jmsTemplate");
        Destination queueDestination=(Destination) context.getBean("queueDestination");
        jmsTemplate.send(queueDestination, new MessageCreator(){
            @Override
            public Message createMessage(Session session) throws JMSException {
                return session.createTextMessage("Hello spring JMS");
            }
        });
    }
}

消費者：

package com.jalja.org.jms.spring;

import javax.jms.Destination;

import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.jms.core.JmsTemplate;

public class SpringJmsReceive {
public static void main(String[] args) {
    ApplicationContext context=new ClassPathXmlApplicationContext("spring-jms.xml");
    JmsTemplate jmsTemplate=(JmsTemplate) context.getBean("jmsTemplate");
    Destination queueDestination=(Destination) context.getBean("queueDestination");
    String msg=(String) jmsTemplate.receiveAndConvert(queueDestination);
    System.out.println(msg);
　　}
}

二、PTP的非同步調用

 我們在spring中直接配置非同步接收消息的監聽器，這樣就相當於在spring中配置了消費者，在接受消息的時候就不必要啟動消費者了。

spring-jms.xml:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:jms="http://www.springframework.org/schema/jms"
xmlns:amq="http://activemq.apache.org/schema/core"
xsi:schemaLocation="http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core.xsd
http://www.springframework.org/schema/jms
http://www.springframework.org/schema/jms/spring-jms.xsd
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd">

<!-- ActiveMQConnectionFactory就是JMS中負責創建到ActiveMQ連接的工廠類 -->
<bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory" >
     <property name="brokerURL" value="tcp://192.168.0.224:61616"/>  
</bean>
<!-- 創建連接池 -->
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">  
    <property name="connectionFactory" ref="connectionFactory"/>  
    <property name="maxConnections" value="10"/>  
</bean>  
<!-- Spring為我們提供了多個ConnectionFactory，有SingleConnectionFactory和CachingConnectionFactory -->
<bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">  
    <property name="targetConnectionFactory" ref="pooledConnectionFactory"/>  
</bean> 

<!-- Spring提供的JMS工具類，它可以進行消息發送、接收等 -->  
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">  
    <!-- 這個connectionFactory對應的是我們定義的Spring提供的那個ConnectionFactory對象 -->  
    <property name="connectionFactory" ref="cachingConnectionFactory"/>  
</bean> 

<!--這個是隊列目的地，點對點的-->  
<bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">  
    <constructor-arg index="0" value="spring-queue"/> 
</bean>   

<!-- 消息監聽器 -->  
<bean id="myMessageListener" class="com.jalja.org.jms.spring.yb.MyMessageListener"/>  

<!-- 消息監聽容器 -->  
<bean id="jmsContainer"  class="org.springframework.jms.listener.DefaultMessageListenerContainer">  
    <property name="connectionFactory" ref="cachingConnectionFactory" />  
    <property name="destination" ref="queueDestination" />  
    <property name="messageListener" ref="myMessageListener" />  
</bean> 
</beans>

　　生產者往指定目的地Destination發送消息後，接下來就是消費者對指定目的地的消息進行消費了。那麼消費者是如何知道有生產者發送消息到指定目的地Destination了呢？這是通過Spring為我們封裝的消息監聽容器MessageListenerContainer實現的，它負責接收信息，並把接收到的信息分發給真正的MessageListener進行處理。每個消費者對應每個目的地都需要有對應的MessageListenerContainer。對於消息監聽容器而言，除了要知道監聽哪個目的地之外，還需要知道到哪裡去監聽，也就是說它還需要知道去監聽哪個JMS伺服器，這是通過在配置MessageConnectionFactory的時候往裡面註入一個ConnectionFactory來實現的。所以我們在配置一個MessageListenerContainer的時候有三個屬性必須指定，一個是表示從哪裡監聽的ConnectionFactory；一個是表示監聽什麼的Destination；一個是接收到消息以後進行消息處理的MessageListener。Spring一共為我們提供了兩種類型的MessageListenerContainer，SimpleMessageListenerContainer和DefaultMessageListenerContainer。
　　SimpleMessageListenerContainer：SimpleMessageListenerContainer會在一開始的時候就創建一個會話session和消費者Consumer，並且會使用標準的JMS MessageConsumer.setMessageListener()方法註冊監聽器讓JMS提供者調用監聽器的回調函數。它不會動態的適應運行時需要和參與外部的事務管理。相容性方面，它非常接近於獨立的JMS規範，但一般不相容Java EE的JMS限制。

　　DefaultMessageListenerContainer：在大多數情況下我們還是使用的DefaultMessageListenerContainer，跟SimpleMessageListenerContainer相比，DefaultMessageListenerContainer會動態的適應運行時需要，並且能夠參與外部的事務管理。它很好的平衡了對JMS提供者要求低、先進功能如事務參與和相容Java EE環境。

消息生產者：

    public static void main(String[] args) {
        ApplicationContext context=new ClassPathXmlApplicationContext("spring-jms.xml");
        JmsTemplate jmsTemplate=(JmsTemplate) context.getBean("jmsTemplate");
        Destination queueDestination=(Destination) context.getBean("queueDestination");
        System.out.println("非同步調用執行開始");
        jmsTemplate.send(queueDestination, new MessageCreator(){
            @Override
            public Message createMessage(Session session) throws JMSException {
                return session.createTextMessage("Hello spring JMS");
            }
        });
        System.out.println("非同步調用執行結束");
    }

消息監聽器：MyMessageListener

public class MyMessageListener implements MessageListener{
    @Override
    public void onMessage(Message message) {
        TextMessage msg= (TextMessage) message;
        try {
            System.out.println("你好："+msg.getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

啟動消息生產者  監聽器的執行結果是：

非同步調用執行開始
非同步調用執行結束
你好：Hello spring JMS

三、發佈訂閱 同步接收

spring-jms.xml:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:jms="http://www.springframework.org/schema/jms"
xmlns:amq="http://activemq.apache.org/schema/core"
xsi:schemaLocation="http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core.xsd
http://www.springframework.org/schema/jms
http://www.springframework.org/schema/jms/spring-jms.xsd
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd">

<!-- ActiveMQConnectionFactory就是JMS中負責創建到ActiveMQ連接的工廠類 -->
<bean id="connectionFactory" class="org.apache.activemq.spring.ActiveMQConnectionFactory" >
     <property name="brokerURL" value="tcp://192.168.0.224:61616"/>  
</bean>
<!-- 創建連接池 -->
<bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory" destroy-method="stop">  
    <property name="connectionFactory" ref="connectionFactory"/>  
    <property name="maxConnections" value="10"/>  
</bean>  
<!-- Spring為我們提供了多個ConnectionFactory，有SingleConnectionFactory和CachingConnectionFactory -->
<bean id="cachingConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">  
    <property name="targetConnectionFactory" ref="pooledConnectionFactory"/>  
</bean> 

<!-- Spring提供的JMS工具類，它可以進行消息發送、接收等 -->  
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">  
    <!-- 這個connectionFactory對應的是我們定義的Spring提供的那個ConnectionFactory對象 -->  
    <property name="connectionFactory" ref="cachingConnectionFactory"/>  
</bean> 

<!--這個是隊列目的地，發佈訂閱-->  
<bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">  
    <constructor-arg index="0" value="spring-Topic"/> 
</bean>   
</beans>

生產者：

    public static void main(String[] args) {
        ApplicationContext context=new ClassPathXmlApplicationContext("spring-jms.xml");
        JmsTemplate jmsTemplate=(JmsTemplate) context.getBean("jmsTemplate");
        Destination topicDestination=(Destination) context.getBean("topicDestination");
        jmsTemplate.send(topicDestination, new MessageCreator(){
            @Override
            public Message createMessage(Session session) throws JMSException {
                return session.createTextMessage("Hello spring JMS topicDestination");
            }
        });
    }

消費者：

public class SpringJmsSubscriber {
    public static void main(String[] args) {
        ApplicationContext context=new ClassPathXmlApplicationContext("spring-jms.xml");
        JmsTemplate jmsTemplate=(JmsTemplate) context.getBean("jmsTemplate");
        Destination topicDestination=(Destination) context.getBean("topicDestination");
        String msg=(String) jmsTemplate.receiveAndConvert(topicDestination);
        System.out.println(msg);
    }
}

 


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • WPF: 實現 ScrollViewer 滾動到指定控制項處
    在前端 UI 開發中，有時，我們會遇到這樣的需求：在一個 ScrollViewer 中有很多內容，而我們需要實現在執行某個操作後能夠定位到其中指定的控制項處；這很像在 HTML 頁面中點擊一個鏈接後定位到當前網頁上的某個 anchor。 要實現它，首先我們需要看 ScrollViewer 為我們提供的 ...
  • csharp: Use of Is and As operators in csharp
    TypeInfo，PropertyInfo，MethodInfo，FieldInfo ...
  • C#中==運算符
    在這篇博客中，我們將介紹如下內容： ==運算符與基元類型 ==運算符與引用類型 ==運算符與String類型 ==運算符與值類型 ==運算符與泛型 ==運算符與基元類型 我們分別用兩種方式比較兩個整數，第一個使用的是Equals(int)方法，每二個使用的是==運算符： 運行上面的示例，兩個語句出的 ...
  • 自定義分散式RESTful API鑒權機制
    微軟利用OAuth2為RESTful API提供了完整的鑒權機制，但是可能微軟保姆做的太完整了，在這個機制中指定了數據持久化的方法是用EF，而且對於用戶、許可權等已經進行了封裝，對於系統中已經有了自己的用戶表，和不是採用EF做持久化的系統來說限制太大，不自由，而且現實中很多情況下，授權伺服器和資源服務 ...
  • 背水一戰 Windows 10 (55) - 控制項（集合類）: ItemsControl - SemanticZoom, ISemanticZoomInformation
    背水一戰 Windows 10 之 控制項（集合類 - ItemsControl）: SemanticZoom, ISemanticZoomInformation ...
  • 打造獨立資料庫訪問的中間服務
    隨著公司業務的不斷變化，幾年前的 A 項目和底層 DB_A 資料庫華麗轉身為核心業務服務和核心資料庫。 想從 DB_A 資料庫獲取數據的 web 服務越來越多，項目之間的關係逐漸演變為下麵這樣： 很容易看出來按上圖這樣的發展趨勢會存在很多問題（項目關係為個人抽象出來的簡化版，實際情況比這要複雜的多） ...
  • 再談AbstractQueuedSynchronizer：獨占模式
    關於AbstractQueuedSynchronizer JDK1.5之後引入了併發包java.util.concurrent，大大提高了Java程式的併發性能。關於java.util.concurrent包我總結如下： AbstractQueuedSynchronizer是併發類諸如Reentra ...
  • IO【轉換流,列印流,序列化】
    編碼: 把看的懂,變成看不懂的 String str = "中國"; byte[] bytes = str.getBytes(); System.out.println(Arrays.toString(bytes));解碼: 把看不懂的內容,變成能看懂的 String s = new String( ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類