碼上快樂

相關內容    簡體    繁體

ssm整合ActiveMQ

本文轉載自   查看原文   2017-11-20 22:53   2924 

環境：windows7

本次使用spring+springmvc+mybatis整合activemq

整合結果，項目能正常運行。並且使用postman進行接口測試，沒有進行數據的持久化到數據庫，只是引入了一個properties文件，原本應該還引入數據庫配置文件的，此處省略，如果要引入，需要注意spring引入多個properties文件的問題。按照本文的引入方法直接引入多個properties文件可行。

以下是整合內容

===============================================================================

1，pom.xml文件下載activemq的jar包配置，僅貼出activemq的配置

        <!--activemq  -->

　　<dependency>
　　　　<groupId>org.springframework</groupId>
　　　　<artifactId>spring-jms</artifactId>
　　　　<version>${spring.version}</version>
　　</dependency>
　　<dependency>
　　　　<groupId>org.apache.activemq</groupId>
　　　　<artifactId>activemq-core</artifactId>
　　　　<version>5.7.0</version>
　　</dependency>
　　<dependency>
　　　　<groupId>org.apache.activemq</groupId>
　　　　<artifactId>activemq-pool</artifactId>
　　　　<version>5.12.2</version>
　　</dependency>

        <!-- xbean 如<amq:connectionFactory /> 此次配置主要是因為配置文件中可能讀取不到amq標簽 -->
    <dependency>
      <groupId>org.apache.xbean</groupId>
      <artifactId>xbean-spring</artifactId>
      <version>4.5</version>
    </dependency>

============================================================================

2，spring的配置

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:aop="http://www.springframework.org/schema/aop"
    xmlns:mvc="http://www.springframework.org/schema/mvc" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:context="http://www.springframework.org/schema/context" xmlns:tx="http://www.springframework.org/schema/tx"
    xmlns:task="http://www.springframework.org/schema/task" xmlns:jdbc="http://www.springframework.org/schema/jdbc"
    xsi:schemaLocation="
        http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop.xsd
        http://www.springframework.org/schema/mvc http://www.springframework.org/schema/mvc/spring-mvc.xsd
        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
        http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd
        http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
        http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd">
<bean class="org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor" />


<mvc:default-servlet-handler />
<mvc:annotation-driven /><!-- 必須添加mvc:annotation-driven 否則所有controller失效 -->
<!-- 掃描注解Bean -->
<context:annotation-config></context:annotation-config>
<context:component-scan base-package="com.jjmc.client.controller" />
<context:component-scan base-package="com.jjmc.client.service" />
<context:component-scan base-package="com.jjmc.client.service.impl" />
<!--     指定Sping組件掃描的基本包路徑
    <context:component-scan base-package="com.jjmc.client" >
        這里只掃描Controller，不可重復加載Service
        <context:include-filter type="annotation" expression="org.springframework.stereotype.Controller"/>
    </context:component-scan> -->
<!--開啟這個配置，spring才能識別@Scheduled注解-->  
    <task:annotation-driven/> 
    
    
</beans>

=====================================================================================

3，activemq的配置文件

<?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:amq="http://activemq.apache.org/schema/core"
       xmlns:jms="http://www.springframework.org/schema/jms"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:mvc="http://www.springframework.org/schema/mvc"
       xsi:schemaLocation="
        http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/context
        http://www.springframework.org/schema/context/spring-context.xsd
        http://www.springframework.org/schema/mvc
        http://www.springframework.org/schema/mvc/spring-mvc.xsd
        http://www.springframework.org/schema/jms
        http://www.springframework.org/schema/jms/spring-jms.xsd
        http://activemq.apache.org/schema/core
        http://activemq.apache.org/schema/core/activemq-core-5.12.2.xsd"
> 

<!-- 加載activemq配置文件 -->
<context:property-placeholder location="classpath:activemq/activemq.properties"
ignore-unresolvable="true" />

<!-- 使用amq標簽啟動過慢 -->
<!-- <amq:connectionFactory id="amqConnectionFactory"
brokerURL="tcp://127.0.0.1:61616"
userName="admin"
password="admin" /> -->

<!--更新一下，配置可以寫在properties文件中-->
<!--<bean id="amqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://127.0.0.1:61616" />
<property name="userName" value="admin" />
<property name="password" value="admin" />
<!-- 是否異步發送消息 -->
<property name="useAsyncSend" value="true" />
<property name="alwaysSessionAsync" value="false" />
</bean>-->

<!--更新一下，配置可以寫在properties文件中-->
<bean id="amqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="${activemq.connectType}://${activemq.brokerURL}:${activemq.port}" />
<property name="userName" value="${activemq.username}" />
<property name="password" value="${activemq.password}" />
<!-- 是否異步發送消息 -->
<property name="useAsyncSend" value="true" />
<property name="alwaysSessionAsync" value="false" />
</bean>

 

<!-- 設置發送連接池,提高性能 -->
<bean id="JmsSenderFactory" class="org.apache.activemq.pool.PooledConnectionFactory"
destroy-method="stop">
<property name="connectionFactory" ref="amqConnectionFactory" />
<property name="maxConnections" value="20"></property>
</bean>

<!-- 設置接收連接池,提高性能 -->
<bean id="JmsReveiverFactory" class="org.apache.activemq.pool.PooledConnectionFactory">
<property name="connectionFactory" ref="amqConnectionFactory" />
<property name="maxConnections" value="10"></property>
</bean>



    <!-- 消息傳輸監聽器 處理網絡及服務器異常 -->
<!--     <property name="transportListener">
        <bean class="cn.mayongfa.activemq.ActiveMQTransportListener" />
    </property> -->


    <!-- 配置JMS連接工長 -->
    <bean id="connectionFactory"
          class="org.springframework.jms.connection.CachingConnectionFactory">
        <constructor-arg ref="amqConnectionFactory" />
        <property name="sessionCacheSize" value="100" />
    </bean>

<!--因為寫在配置文件中所以-->
    <!-- 定義消息隊列（Queue） -->
<!--    <bean id="demoQueueDestination" class="org.apache.activemq.command.ActiveMQQueue">
        <!-- 設置消息隊列的名字 -->
        <constructor-arg>
            <value>ActiveMQ</value>
        </constructor-arg>
    </bean>-->

 <!-- 定義消息隊列（Queue） -->

<bean id="demoQueueDestination" class="org.apache.activemq.command.ActiveMQQueue">
<!-- 設置消息隊列的名字 -->
　　<constructor-arg>
　　　　<value>${activemq.virtual.host}</value>
　　</constructor-arg>
</bean>



    <!-- 配置JMS模板（Queue），Spring提供的JMS工具類，它發送、接收消息。 -->
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="defaultDestination" ref="demoQueueDestination" />
        <!--JmsTemplate的接收是阻塞式的接收，默認會一直阻塞等待，直到接收到了消息。也可以設置一個最長的等待時間參數，超過這個時間，接收的方法將得到null的結果。  -->
        <property name="receiveTimeout" value="10000" />
        <!-- true是topic，false是queue，默認是false，此處顯示寫出false。非pub/sub模型（發布/訂閱），即隊列模式 -->
        <property name="pubSubDomain" value="false" />
        <!-- deliveryMode, priority, timeToLive 的開關，要生效，必須配置為true，默認false-->  
        <property name="explicitQosEnabled" value="true" /> 
        <!-- 發送模式  DeliveryMode.NON_PERSISTENT=1:非持久 ; DeliveryMode.PERSISTENT=2:持久-->
        <property name="deliveryMode" value="1" />
        <!--在發送時，可以設置消息的存活時間，默認為永久有效。  -->
        <!-- <property name="timeToLive" value="5000" /> --><!--此項配置有問題  -->
    <!-- 消息應答方式  
        Session.AUTO_ACKNOWLEDGE  消息自動簽收  
        Session.CLIENT_ACKNOWLEDGE  客戶端調用acknowledge方法手動簽收  
        Session.DUPS_OK_ACKNOWLEDGE 不必必須簽收，消息可能會重復發送 
        JMS規范的ack消息確認機制有一下四種，定於在session對象中：
AUTO_ACKNOWLEDGE = 1 ：自動確認
CLIENT_ACKNOWLEDGE = 2：客戶端手動確認 
DUPS_OK_ACKNOWLEDGE = 3： 自動批量確認
SESSION_TRANSACTED = 0：事務提交並確認
但是在activemq補充了一個自定義的ACK模式:
INDIVIDUAL_ACKNOWLEDGE = 4：單條消息確認 
        -->     
    </bean>




   <!--  配置消息隊列監聽者（Queue） -->
    <bean id="queueMessageListener" class="com.jjmc.client.filter.QueueMessageListener" />


<!--     顯示注入消息監聽容器（Queue），配置連接工廠，監聽的目標是demoQueueDestination，監聽器是上面定義的監聽器 -->
    <bean id="queueListenerContainer"
          class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destination" ref="demoQueueDestination" />
        <property name="messageListener" ref="queueMessageListener" />
    </bean>


</beans>

 

===================================================================================

4,web.xml的配置文件

<web-app xmlns="http://java.sun.com/xml/ns/javaee"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_3_0.xsd"
    version="3.0">
    
  <display-name>business-client</display-name>


  <welcome-file-list>
        <welcome-file>/index.jsp</welcome-file>
    </welcome-file-list>


    <servlet>
        <servlet-name>business-client</servlet-name>
        <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
        <init-param>
            <param-name>contextConfigLocation</param-name>
            <param-value></param-value>
        </init-param>
        <load-on-startup>1</load-on-startup>
    </servlet>


<!-- 設置spring訪問所有資源 -->
    <servlet-mapping>
        <servlet-name>business-client</servlet-name>
        <url-pattern>/</url-pattern>
    </servlet-mapping>


<!-- Spring Config -->
<context-param>
<param-name>contextConfigLocation</param-name>
<param-value>classpath:spring/applicationContext.xml;classpath:spring/applicationContext-ActiveMQ.xml</param-value>
</context-param>
<listener>
   <listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>
</listener>


    <!-- Spring刷新Interceptor防止內存泄漏  -->
    <listener>
        <listener-class>org.springframework.web.util.IntrospectorCleanupListener</listener-class>
    </listener>


<!-- 設置session 超時時間為20分鍾 -->
<session-config>
<session-timeout>20</session-timeout>
</session-config>
  <!-- 處理編碼格式 -->
  <filter>
    <filter-name>characterEncodingFilter</filter-name>
    <filter-class>org.springframework.web.filter.CharacterEncodingFilter</filter-class>
    <init-param>
      <param-name>encoding</param-name>
      <param-value>UTF-8</param-value>
    </init-param>
    <init-param>
      <param-name>forceEncoding</param-name>
      <param-value>true</param-value>
    </init-param>
  </filter>
  <filter-mapping>
    <filter-name>characterEncodingFilter</filter-name>
    <url-pattern>/*</url-pattern>
  </filter-mapping>
</web-app>

=====================================================================

5，controller層

 

/**
 * 2017年11月20日下午8:09:48
 */
package com.jjmc.client.controller;


import javax.annotation.Resource;
import javax.jms.Destination;
import javax.jms.TextMessage;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;


import org.apache.log4j.Logger;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;


import com.jjmc.client.service.ConsumerService;
import com.jjmc.client.service.ProducerService;


/**
 *
 * @author huangtao
 * 2017年11月20日下午8:09:48
 * business-client
 * @parameter
 * TODO
 *
 */
@Controller
public class MessageController {
    private Logger logger = Logger.getLogger(MessageController.class);
    @Resource(name = "demoQueueDestination")
    private Destination destination;


    //隊列消息生產者
    @Resource(name = "producerService")
    private ProducerService producer;


    //隊列消息消費者
    @Resource(name = "consumerService")
    private ConsumerService consumer;


    @RequestMapping(value = "/SendMessage", method = RequestMethod.POST)
    @ResponseBody
    public void send(HttpServletRequest request, HttpServletResponse response,@RequestBody String msg) {
        logger.info(Thread.currentThread().getName()+"------------send to jms Start");
        producer.sendMessage(msg);
        logger.info(Thread.currentThread().getName()+"------------send to jms End");
    }


    @RequestMapping(value= "/ReceiveMessage",method = RequestMethod.GET)
    @ResponseBody
    public Object receive(){
        logger.info(Thread.currentThread().getName()+"------------receive from jms Start");
        TextMessage tm = consumer.receive(destination);
        logger.info(Thread.currentThread().getName()+"------------receive from jms End");
        return tm;
    }


}

 

===================================================================

6，生產者service

/**
 * 2017年11月20日下午8:08:24
 */
package com.jjmc.client.service;


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


import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Service;


/**
 *
 * @author huangtao
 * 2017年11月20日下午8:08:24
 * business-client
 * @parameter
 * TODO
 *
 */
@Service
public class ProducerService {


    @Resource(name="jmsTemplate")
    private JmsTemplate jmsTemplate;


    public void sendMessage(Destination destination,String msg){
        System.out.println(Thread.currentThread().getName()+" 向隊列"+destination.toString()+"發送消息---------------------->"+msg);
        jmsTemplate.send(destination, new MessageCreator() {
            public Message createMessage(Session session) throws JMSException {
                return session.createTextMessage(msg);
            }
        });
    }


    public void sendMessage(String msg){
        String destination = jmsTemplate.getDefaultDestinationName();
        System.out.println(Thread.currentThread().getName()+" 向隊列"+destination+"發送消息---------------------->"+msg);
        jmsTemplate.send(new MessageCreator() {
            public Message createMessage(Session session) throws JMSException {
                return session.createTextMessage(msg);
            }
        });
    }
}

 

========================================================================

7,消費者service

/**
 * 2017年11月20日下午8:09:12
 */
package com.jjmc.client.service;


import javax.annotation.Resource;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.TextMessage;


import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service;


/**
 *
 * @author huangtao
 * 2017年11月20日下午8:09:12
 * business-client
 * @parameter
 * TODO
 *
 */
@Service
public class ConsumerService {
    @Resource(name="jmsTemplate")
    private JmsTemplate jmsTemplate;


    public TextMessage receive(Destination destination){
        TextMessage textMessage = (TextMessage) jmsTemplate.receive(destination);
        try{
            System.out.println("從隊列" + destination.toString() + "收到了消息：\t"
                    + textMessage.getText());
        } catch (JMSException e) {
            e.printStackTrace();
        }
        return textMessage;
    }
}

======================================================================

8,監聽器filter

注意：關於監聽器，需要注意，如果配置了監聽器，消息隊列里的消息放入就會立即被監聽到，就會自動被取出，不配置監聽器，就可以調用取出的方法

/**
 * 2017年11月20日下午10:00:27
 */
package com.jjmc.client.filter;


import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;


/**
 *
 * @author huangtao
 * 2017年11月20日下午10:00:27
 * business-client
 * @parameter
 * TODO
 *
 */
public class QueueMessageListener implements MessageListener {
    public void onMessage(Message message) {
        TextMessage tm = (TextMessage) message;
        try {
            System.out.println("QueueMessageListener監聽到了文本消息：\t"
                    + tm.getText());
            //do something ...
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}

 

=============================================================

9,增加一個activemq.properties

#rabbitmq
activemq.brokerURL=127.0.0.1
activemq.port=61616
activemq.connectType=tcp
activemq.username=admin
activemq.password=admin

activemq.virtual.host=ActiveMQ

 ============================================================

 


免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 SSM整合 SSM整合 SSM整合 SSM 整合 SSM整合 SSM整合 springboot與ActiveMQ整合 SpringBoot 整合 ActiveMq springboot整合activeMQ SpringBoot整合ActiveMQ
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM