JMS结合ActiveMQ实现消息传递

以你之姓@ 2022-03-27 01:52 271阅读 0赞

> ### JMS结合ActiveMQ实现消息传递 ###

**【定义概念】：**[ 参考文献：java消息服务使用总结（JMS：通过ActiveMQ实现)][_java_JMS_ActiveMQ]

**【JDBC-持久化方式】 (一种ActiveMQ消息持久化机制)：**

**修改/conf/activemq.xml配置文件，****首先定义一个mysql-ds的MySQL数据源，然后在persistenceAdapter节点中配置jdbcPersistenceAdapter并且引用刚才定义的数据源。dataSource指定持久化数据库的bean，createTablesOnStartup是否在启动的时候创建数据表，默认值是true，这样每次启动都会去创建数据表了，一般是第一次启动的时候设置为true，之后改成false。**

**在broker节点后添加数据源配置：**

</broker>
    	
    <bean id="mysql-ds" class="org.apache.commons.dbcp2.BasicDataSource" destroy-method="close"> 
    	<property name="driverClassName" value="com.mysql.jdbc.Driver"/> 
    	<property name="url" value="jdbc:mysql://localhost/activemq"/>
    	<property name="username" value="root"/> 
    	<property name="password" value="123456"/> 
    	<property name="poolPreparedStatements" value="true"/> 
    </bean>

【说明】： **上述数据源中的 class="org.apache.commons.dbcp2.BasicDataSource"配置是 本实例运用activemq版本是5.15.8（高版本）；低版本的MQ应该是 org.apache.commons.dbcp.BasicDataSource；最后将mysql驱动jar包放入apache-activemq-5.15.8/lib下即可**

**首次启动后就会看到数据库的创建的三张表：**

![20190111171033224.png][]

**【pom依赖】：**

<dependency>
          <groupId>org.apache.activemq</groupId>
          <artifactId>activemq-all</artifactId>
          <version>5.14.1</version>
        </dependency>
        <dependency>
          <groupId>org.springframework</groupId>
          <artifactId>spring-jms</artifactId>
          <version>4.0.8.RELEASE</version>
        </dependency>
        <dependency>
          <groupId>org.springframework</groupId>
          <artifactId>spring-messaging</artifactId>
          <version>4.0.8.RELEASE</version>
        </dependency>
        <dependency>
          <groupId>org.apache.activemq</groupId>
          <artifactId>activemq-spring</artifactId>
          <version>5.14.1</version>
        </dependency>

**【spring-activemq.xml】 配置：****【待解决】：发现activemq配置启动和rabbitmq配置ConnectionFactory冲突，暂时注释掉启动配置中的rabbitmq配置即可**

<?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"
           xsi:schemaLocation="http://www.springframework.org/schema/beans
            http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
            http://www.springframework.org/schema/jms
            http://www.springframework.org/schema/jms/spring-jms-4.0.xsd
            http://activemq.apache.org/schema/core
            http://activemq.apache.org/schema/core/activemq-core-5.14.1.xsd">
        <description>spring-activeMQ</description>
        
        <amq:connectionFactory id="amqConnectionFactory" brokerURL="tcp://127.0.0.1:61616" userName="admin" password="admin"></amq:connectionFactory>
    
        
        <bean id="activeMQConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
            <constructor-arg ref="amqConnectionFactory"></constructor-arg>
            <property name="sessionCacheSize" value="100" />
        </bean>
    
        
        <bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">
            
            <constructor-arg ref="activeMQConnectionFactory"></constructor-arg>
            
            <property name="pubSubDomain" value="false" />
        </bean>
    
        
        <bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate">
            
            <constructor-arg ref="activeMQConnectionFactory"></constructor-arg>
            
            <property name="pubSubDomain" value="true" />
        </bean>
    
    
        <bean id="queueReceiver" class="com.caox.activemq.annotation.queue.QueueReceiver"></bean>
        
        <jms:listener-container destination-type="queue" container-type="default" connection-factory="activeMQConnectionFactory" acknowledge="auto">
            <jms:listener destination="test.queue" ref="queueReceiver"/>
        </jms:listener-container>
    
        <bean id="topicReceiver" class="com.caox.activemq.annotation.topic.TopicReceiver"></bean>
        
        <jms:listener-container destination-type="topic" container-type="default" connection-factory="activeMQConnectionFactory" acknowledge="auto">
            <jms:listener destination="test.topic" ref="topicReceiver"/>
        </jms:listener-container>
    </beans>

**【spirng配置监听消息】 -【queue发送和消费服务】：**

**  (1)生产者：**

package com.caox.activemq.annotation.queue;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.jms.core.MessageCreator;
    import org.springframework.stereotype.Component;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.Session;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/1/11 14:28
     */
    @Component("queueSender")
    @Slf4j
    public class QueueSender {
        @Autowired
        @Qualifier("jmsQueueTemplate")
        private JmsTemplate jmsTemplate;
    
        public void send(String queueName, final String message){
            log.info("call 发送 activemq queue 消息 | queueName：{} message:{}",queueName, message);
            jmsTemplate.send(queueName, new MessageCreator() {
                @Override
                public Message createMessage(Session session) throws JMSException {
                    return session.createTextMessage(message);
                }
            });
        }
    }

**(2)消费者：**

package com.caox.activemq.annotation.queue;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/1/11 14:22
     */
    public class QueueReceiver implements MessageListener {
            @Override
            public void onMessage(Message message) {
                try {
                    System.out.println("QueueReceiver接收到消息："+((TextMessage)message).getText());
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        }

**【spirng配置监听消息】 -【topic发送和消费服务】：**

**(1)生产者：**

package com.caox.activemq.annotation.topic;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Qualifier;
    import org.springframework.jms.core.JmsTemplate;
    import org.springframework.jms.core.MessageCreator;
    import org.springframework.stereotype.Component;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.Session;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/1/11 14:30
     */
    @Component("topicSender")
    @Slf4j
    public class TopicSender{
        @Autowired
        @Qualifier("jmsTopicTemplate")
        private JmsTemplate jmsTemplate;
        public void send(String queueName, final String message){
            log.info("call 发送 activemq topic 消息 | queueName：{} message:{}",queueName, message);
            jmsTemplate.send(queueName,new MessageCreator() {
                @Override
                public Message createMessage(Session session) throws JMSException {
                    return session.createTextMessage(message);
                }
            });
        }
    }

**(2)消费者：**

package com.caox.activemq.annotation.topic;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    /**
     * @author : nazi
     * @version : 1.0
     * @date : 2019/1/11 14:32
     */
    public class TopicReceiver implements MessageListener{
        @Override
        public void onMessage(Message message) {
            try {
                System.out.println("TopicReceiver接收到消息："+((TextMessage)message).getText());
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }

[【参考文献】：ActiveMQ的几种消息持久化机制 ][ActiveMQ]

[_java_JMS_ActiveMQ]: https://blog.csdn.net/qq30211478/article/details/78599815
[20190111171033224.png]: /images/20220327/d0fe5781d5fa4d81a6ccfb16c6047150.png
[ActiveMQ]: https://www.cnblogs.com/binyue/p/5371479.html