JMS 之 Active MQ 的spring整合

阿新 • • 發佈：2017-06-23

rop tin eat code producer 什麽我們 const cati

一、與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);
    }
}

JMS 之 Active MQ 的spring整合

rop tin eat code producer 什麽我們 const cati 一、與spring整合實現ptp的同步接收消息 pom.xml: <!-- https://mvnrepository.com/artifact/org.springfram

JMS 之 Active MQ 的spring整合

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

二、PTP的異步調用

三、發布訂閱同步接收

JMS 之 Active MQ 的spring整合

JMS 之 Active MQ 啟動嵌入式Broke

淺談jms之（通過spring整合activeMQ實現jms）例項

jms之activeMQ與spring整合進階-實現一種負載均衡

Spring整合JMS、IBM MQ發送和接收消息

active mq 整合spring 並持久化到資料庫

Spring Cloud Edgware新特性之九：Sleuth使用MQ方式整合Zipkin

Spring整合JMS、IBM MQ傳送和接收訊息

spring整合jms

Spring整合JMS(消息中間件)

消息中間件ActiveMQ及Spring整合JMS的介紹

Mybatis 和Spring整合之原始dao開發

Mybatis 和Spring整合之mapper代理開發

【SSH框架】系列之 Spring 整合 Hibernate 框架

07-spring之三大框架的整合

spring boot 系列之三：spring boot 整合JdbcTemplate

spring boot 系列之四：spring boot 整合JPA

Spring整合Struts2和Hibernate+Maven（三）之請求的處理

Spring整合Struts2和Hibernate+Maven（二）之SSH的配置檔案

Spring整合Struts2和Hibernate+Maven（一）之Maven專案建立

JMS 之 Active MQ 的spring整合

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

二、PTP的異步調用

三、發布訂閱 同步接收

相關推薦

三、發布訂閱同步接收