ActiveMQ+Spring 工程创建详解

柔情只为你懂 2022-07-13 06:57 198阅读 0赞

ActiveMQ是Apache所提供的一个开源的消息系统，完全采用Java来实现，因此，它能很好地支持J2EE提出的JMS Java Message Service,即Java消息服务）规范。JMS是一组Java应用程序接口，它提供消息的创建、发送、读取等一系列服务。JMS提供了一组公共应用程序接口和响应的语法，类似于Java数据库的统一访问接口JDBC,它是一种与厂商无关的API，使得Java程序能够与不同厂商的消息组件很好地进行通信。

JMS支持两种消息发送和接收模型。一种称为P2P(Ponit to Point)模型，即采用点对点的方式发送消息。P2P模型是基于队列的，消息生产者发送消息到队列，消息消费者从队列中接收消息，队列的存在使得消息的异步传输称为可能，P2P模型在点对点的情况下进行消息传递时采用。

![这里写图片描述][20160519152911514]

另一种称为Pub/Sub(Publish/Subscribe，即发布-订阅)模型，发布-订阅模型定义了如何向一个内容节点发布和订阅消息，这个内容节点称为topic(主题)。主题可以认为是消息传递的中介，消息发布这将消息发布到某个主题，而消息订阅者则从主题订阅消息。主题使得消息的订阅者与消息的发布者互相保持独立，不需要进行接触即可保证消息的传递，发布-订阅模型在消息的一对多广播时采用。

![这里写图片描述][20160519152926893]

### ActiveMQ的安装 ###

下载最新的安装包apache-activemq-5.13.2-bin.tar.gz

下载之后解压： tar -zvxf apache-activemq-5.13.2-bin.tar.gz

**ActiveMQ目录内容有：**

*  bin目录包含ActiveMQ的启动脚本
 *  conf目录包含ActiveMQ的所有配置文件
 *  data目录包含日志文件和持久性消息数据
 *  example: ActiveMQ的示例
 *  lib: ActiveMQ运行所需要的lib
 *  webapps: ActiveMQ的web控制台和一些相关的demo

ActiveMQ的默认服务端口为61616，这个可以在conf/activemq.xml配置文件中修改：

<transportConnectors>
        
        <transportConnector name="openwire" uri="tcp://0.0.0.0:61616?maximumConnections=1000&wireFormat.maxFrameSize=104857600"/>
    </transportConnectors>
     
      
       1
       2
       3
       4
      
     
      
      
       
     
      
     
      
       1
       2
       3
       4

**JMS的规范流程**

1.  获得JMS connection factory. 通过我们提供特定环境的连接信息来构造factory。
2.  利用factory构造JMS connection
3.  启动connection
4.  通过connection创建JMS session.
5.  指定JMS destination.
6.  创建JMS producer或者创建JMS message并提供destination.
7.  创建JMS consumer或注册JMS message listener.
8.  发送和接收JMS message.
9.  关闭所有JMS资源，包括connection, session, producer, consumer等。

### 案例（整合Spring） ###

`pom.xml`

<dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>4.11</version>
      <scope>test</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.activemq</groupId>
      <artifactId>activemq-all</artifactId>
      <version>5.11.1</version>
    </dependency>
    <dependency>
      <groupId>org.springframework</groupId>
      <artifactId>spring-jms</artifactId>
      <version>4.1.4.RELEASE</version>
    </dependency>
    <dependency>
      <groupId>org.springframework</groupId>
      <artifactId>spring-test</artifactId>
      <version>4.1.4.RELEASE</version>
    </dependency>
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
      
     
      
      
       
     
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21

**Queue类型消息**

`spring配置文件`

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
      
      <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        
        <property name="brokerURL" value="failover:(tcp://192.168.147.131:61616)" />
      </bean>
      
      <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
        
        <constructor-arg>
          <value>testSpringQueue</value>
        </constructor-arg>
      </bean>
    
    
    
    
    
    
      
      <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="defaultDestination" ref="queueDestination" />
        <property name="receiveTimeout" value="10000" />
      </bean>
    </beans>
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28
       29
       30
      
     
      
      
       
     
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28
       29
       30

`推送代码`

package com.mq.spring.queue;
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.jms.core.MessageCreator;
    import org.springframework.test.context.ContextConfiguration;
    import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
    import javax.annotation.Resource;
    import javax.jms.*;
    
    @RunWith(SpringJUnit4ClassRunner.class)
    @ContextConfiguration(locations={
       "classpath:spring-mq-queue.xml"})
    public class QueueSender { 
      @Resource
      private JmsTemplate jmsTemplate;
      @Test
      public void send(){
        sendMqMessage(null,"spring activemq queue type message !");
      }
      /** * 说明:发送的时候如果这里没有显示的指定destination.将用spring xml中配置的destination * @param destination * @param message */
      public void sendMqMessage(Destination destination, final String message){
        if(null == destination){
          destination = jmsTemplate.getDefaultDestination();
        }
        jmsTemplate.send(destination, new MessageCreator() {
          public Message createMessage(Session session) throws JMSException {
            return session.createTextMessage(message);
          }
        });
        System.out.println("spring send message...");
      }
      public void setJmsTemplate(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
      }
    }
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28
       29
       30
       31
       32
       33
       34
       35
       36
       37
       38
       39
      
     
      
      
       
     
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28
       29
       30
       31
       32
       33
       34
       35
       36
       37
       38
       39

`消费代码`

package com.mq.spring.queue;
    import org.junit.Test;
    import javax.jms.*;
    import org.junit.runner.RunWith;
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.jms.core.MessageCreator;
    import org.springframework.test.context.ContextConfiguration;
    import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
    import javax.annotation.Resource;
    import javax.jms.Message;
    
    @RunWith(SpringJUnit4ClassRunner.class)
    @ContextConfiguration(locations={
       "classpath:spring-mq-queue.xml"})
    public class QueueReceiver { 
      @Resource
      private JmsTemplate jmsTemplate;
      @Test
      public void receiveMqMessage(){
        Destination destination = jmsTemplate.getDefaultDestination();
        receive(destination);
      }
      /** * 接受消息 */
      public void receive(Destination destination) {
        TextMessage tm = (TextMessage) jmsTemplate.receive(destination);
        try {
          System.out.println("从队列" + destination.toString() + "收到了消息：\t" + tm.getText());
        } catch (JMSException e) {
          e.printStackTrace();
        }
      }
      public void setJmsTemplate(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
      }
    }
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28
       29
       30
       31
       32
       33
       34
       35
       36
      
     
      
      
       
     
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28
       29
       30
       31
       32
       33
       34
       35
       36

说明:上面的生产者和消费者使用同一套配置文件,使用独立的程序去接收消息,spring jms也提供了消息监听处理.接下来我们换成监听式消费

`配置文件`

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
      
      <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="failover:(tcp://192.168.147.131:61616)" />
      </bean>
      
      <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
        
        <constructor-arg>
          <value>testSpringQueue</value>
        </constructor-arg>
      </bean>
      
      <bean id="consumerMessageListener" class="com.mq.spring.queue.ConsumerMessageListener" />
      
      <bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destination" ref="queueDestination" />
        <property name="messageListener" ref="consumerMessageListener" />
      </bean>
    </beans>
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
      
     
      
      
       
     
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25

`监听器代码`

public class ConsumerMessageListener implements MessageListener{ 
      @Override
      public void onMessage(Message message) {
        TextMessage tm = (TextMessage) message;
        try {
          System.out.println("ConsumerMessageListener收到了文本消息：\t" + tm.getText());
        } catch (JMSException e) {
          e.printStackTrace();
        }
      }
    }
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
      
     
      
      
       
     
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11

这样我们的消息消费就可以在监听器中处理消费了.生产的代码不变,修改发送者的消息体内容,执行生产程序

**Topic类型消息**

在使用 Spring JMS的时候，主题（ Topic）和队列消息的主要差异体现在JmsTemplate中 “pubSubDomain”是否设置为 True。如果为 True，则是 Topic；如果是false或者默认，则是 queue。

<property name="pubSubDomain" value="true" />
     
      
       1
      
     
      
      
       
     
      
     
      
       1

`topic类型消费配置文件说明`

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
      
      <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="failover:(tcp://192.168.147.131:61616)" />
      </bean>
      
      <bean id="destination" class="org.apache.activemq.command.ActiveMQQueue">
        
        <constructor-arg>
          <value>testSpringTopic</value>
        </constructor-arg>
      </bean>
      
      <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="defaultDestination" ref="destination" />
        <property name="receiveTimeout" value="10000" />
      </bean>
      
      <bean id="consumerMessageListener" class="com.mq.spring.topic.ConsumerMessageListener" />
      
      <bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="destination" ref="destination" />
        
        <property name="pubSubDomain" value="true" />
        <property name="messageListener" ref="consumerMessageListener" />
      </bean>
    </beans>
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28
       29
       30
       31
       32
       33
      
     
      
      
       
     
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28
       29
       30
       31
       32
       33

`生产者代码`

package com.mq.spring.topic;
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.jms.core.MessageCreator;
    import org.springframework.test.context.ContextConfiguration;
    import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
    import javax.annotation.Resource;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.Session;
    
    @RunWith(SpringJUnit4ClassRunner.class)
    @ContextConfiguration(locations={
       "classpath:spring-mq-topic.xml"})
    public class TopicSender { 
      @Resource
      private JmsTemplate jmsTemplate;
      @Test
      public void send(){
        sendMqMessage(null,"spring activemq topic type message[with listener] !");
      }
      /** * 说明:发送的时候如果这里没有显示的指定destination.将用spring xml中配置的destination * @param destination * @param message */
      public void sendMqMessage(Destination destination, final String message){
        if(null == destination){
          destination = jmsTemplate.getDefaultDestination();
        }
        jmsTemplate.send(destination, new MessageCreator() {
          public Message createMessage(Session session) throws JMSException {
            return session.createTextMessage(message);
          }
        });
        System.out.println("spring send message...");
      }
      public void setJmsTemplate(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
      }
    }
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28
       29
       30
       31
       32
       33
       34
       35
       36
       37
       38
       39
       40
       41
       42
      
     
      
      
       
     
      
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28
       29
       30
       31
       32
       33
       34
       35
       36
       37
       38
       39
       40
       41
       42

`监听器代码`

参考资料

*    
    
 *  [http://www.importnew.com/19767.html][http_www.importnew.com_19767.html]
 *  [http://blog.csdn.net/u013256816/article/details/51161548][http_blog.csdn.net_u013256816_article_details_51161548]
 *  [http://www.tuicool.com/articles/FVf26b][http_www.tuicool.com_articles_FVf26b]
 *  [http://haohaoxuexi.iteye.com/blog/1893038][http_haohaoxuexi.iteye.com_blog_1893038]
 *  [Spring+ActiveMQ+Mysql 配置JMS][Spring_ActiveMQ_Mysql _JMS]

[20160519152911514]: /images/20220713/85f08d021774465da6e6012c203b940f.png
[20160519152926893]: /images/20220713/ad6dbfe37f1947dba912f8e4fc31877e.png
[http_www.importnew.com_19767.html]: http://www.importnew.com/19767.html
[http_blog.csdn.net_u013256816_article_details_51161548]: http://blog.csdn.net/u013256816/article/details/51161548
[http_www.tuicool.com_articles_FVf26b]: http://www.tuicool.com/articles/FVf26b
[http_haohaoxuexi.iteye.com_blog_1893038]: http://haohaoxuexi.iteye.com/blog/1893038
[Spring_ActiveMQ_Mysql _JMS]: http://blog.csdn.net/wilsonke/article/details/42191015