ActiveMQ（二、代码实现）

冷不防 2022-05-12 14:12 168阅读 0赞

代码实现

**一、springboot+activeMQ**

**1.1 yml配置（部分）**

spring:
      activemq:
        queueName: mingQueue
        topicName: mingTopic
        broker-url: tcp://0.0.0.0:61616
        user:
        password:
        in-memory: true
        pool:
          enabled: false

**1.2 配置类（支持队列和主题）**

package com.ming.activitymq.conf;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.apache.activemq.command.ActiveMQQueue;
    import org.apache.activemq.command.ActiveMQTopic;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.boot.autoconfigure.jms.activemq.ActiveMQProperties;
    import org.springframework.boot.context.properties.EnableConfigurationProperties;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.context.annotation.Primary;
    import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
    import org.springframework.jms.config.JmsListenerContainerFactory;
    import org.springframework.jms.core.JmsMessagingTemplate;
    
    import javax.annotation.Resource;
    import javax.jms.Queue;
    import javax.jms.Topic;
    
    /**
     * 文件名：  ${file_name}
     * 版权：    Copyright by ljm
     * 描述：    activemq简单但配置(支持队列和主题)
     * 修改人：  HuamingChen
     * 修改时间：2018/10/23
     * 跟踪单号：
     * 修改单号：
     * 修改内容：
     */
    @Configuration
    @EnableConfigurationProperties({ActiveMQProperties.class})
    public class MqConfig {
        /**
         * 队列名称
         */
        @Value("${spring.activemq.queueName}")
        private String queueName;
        /**
         * 主题名称
         */
        @Value("${spring.activemq.topicName}")
        private String topicName;
    
        /**
         * 队列
         * @return
         */
        @Bean
        public Queue getQueue(){
            return new ActiveMQQueue(queueName);
        }
    
        /**
         * 主题
         * @return
         */
        @Bean
        public Topic getTopic(){
            return new ActiveMQTopic(topicName);
        }
    
        /**
         * 链接工厂
         * @return
         */
        @Bean
        public ActiveMQConnectionFactory connectionFactory(ActiveMQProperties properties) {
            return new ActiveMQConnectionFactory(properties.getUser(), properties.getPassword(), properties.getBrokerUrl());
        }
    
        /**
         * 主题监听
         * @param connectionFactory
         * @return
         */
        @Bean
        public JmsListenerContainerFactory<?> jmsListenerContainerTopic(ActiveMQConnectionFactory connectionFactory) {
            DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
            bean.setPubSubDomain(true);
            bean.setSessionTransacted(true);
            bean.setConcurrency("3-10");
            bean.setConnectionFactory(connectionFactory);
            return bean;
        }
    
        /**
         * 队列监听
         * @param connectionFactory
         * @return
         */
        @Bean
        public JmsListenerContainerFactory<?> jmsListenerContainerQueue(ActiveMQConnectionFactory connectionFactory) {
            DefaultJmsListenerContainerFactory bean = new DefaultJmsListenerContainerFactory();
            bean.setConnectionFactory(connectionFactory);
            return bean;
        }
    
        /**
         * jms消息模版
         * @param connectionFactory
         * @return
         */
        @Bean
        @Primary
        public JmsMessagingTemplate jmsMessagingTemplate(ActiveMQConnectionFactory connectionFactory){
            return new JmsMessagingTemplate(connectionFactory);
        }
    }

**1.3 简单使用**

package com.ming.activitymq;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.jms.annotation.JmsListener;
    import org.springframework.jms.core.JmsMessagingTemplate;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.ResponseBody;
    import org.springframework.web.bind.annotation.RestController;
    
    import javax.annotation.Resource;
    import javax.jms.Destination;
    import javax.jms.Queue;
    import javax.jms.Topic;
    
    /**
     * 文件名：  ${file_name}
     * 版权：    Copyright by ljm
     * 描述：
     * 修改人：  HuamingChen
     * 修改时间：2018/10/23
     * 跟踪单号：
     * 修改单号：
     * 修改内容：
     */
    @RestController
    @RequestMapping(value="/mq")
    public class ActivityMQController {
        private Logger logger= LoggerFactory.getLogger(ActivityMQController.class);
        @Resource
        private JmsMessagingTemplate jmsMessagingTemplate;
        @Resource
        private Queue queue;
        @Resource
        private Topic topic;
    
        @RequestMapping(value="queueSend")
        @ResponseBody
        public void queueSend(String message) {
            jmsMessagingTemplate.convertAndSend(queue,message);
        }
    
        @RequestMapping(value="topicSend")
        @ResponseBody
        public void topicSend(String message) {
            jmsMessagingTemplate.convertAndSend(topic,message);
        }
    
        @JmsListener(destination = "mingQueue",containerFactory = "jmsListenerContainerQueue")
        public void listenQueueMQ(String message){
            logger.info("**************************队列消息="+message+"************************************");
        }
    
        @JmsListener(destination = "mingTopic",containerFactory = "jmsListenerContainerTopic")
        public void listenTopicMQ(String message){
            logger.info("**************************主题消息"+message+"************************************");
        }
    }

**二、传统方法**

**2.1 基础流程**  
按照JMS的规范，我们首先需要获得一个JMS connection factory.，通过这个connection factory来创建connection.在这个基础之上我们再创建session, destination, producer和consumer。因此主要的几个步骤如下：

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等。

**2.2 创建消费者**

package com.ming.activitymq;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    import javax.jms.*;
    import java.util.UUID;
    
    /**
     * 文件名：  ${file_name}
     * 版权：    Copyright by ljm
     * 描述：    mq提供者
     * 修改人：  HuamingChen
     * 修改时间：2018/10/23
     * 跟踪单号：
     * 修改单号：
     * 修改内容：
     */
    public class MQProduct {
        private static final String USERNAME= ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名
        private static final String PASSWORD= ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码
        private static final String BROKEURL= ActiveMQConnection.DEFAULT_BROKER_URL; // 默认的连接地址
    
        public static void main(String[] args){
            //连接工厂
            ConnectionFactory connectionFactory=null;
            //连接
            Connection connection=null;
            //会话
            Session session=null;
            //目的地
            Destination destination=null;
            //生产者
            MessageProducer messageProducer=null;
            //消息
            TextMessage textMessage=null;
            try {
                //创建连接工厂
                connectionFactory=new ActiveMQConnectionFactory(USERNAME,PASSWORD,BROKEURL);
                //创建连接
                connection=connectionFactory.createConnection();
                //开始连接
                connection.start();
                //创建会话 第一个参数是开启事务true,第二个参数是消息确认机制属性
                session=connection.createSession(true,Session.AUTO_ACKNOWLEDGE);
                //设置目的地
                //destination=session.createQueue("mingTest");//ptp模式
                destination=session.createTopic("mingTestTopic"); //订阅模式
                //创建生产者
                messageProducer=session.createProducer(destination);
                //创建消息
                textMessage=session.createTextMessage("测试消息发送"+ UUID.randomUUID().toString());
                //发送消息
                messageProducer.send(textMessage);
                //提交会话
                session.commit();
                //关闭连接
            } catch (JMSException e) {
                e.printStackTrace();
            } finally {
                if(connection!=null){
                    try {
                        connection.close();
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }

**2.3 创建消费者**

package com.ming.activitymq;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import javax.jms.*;
    
    /**
     * 文件名：  ${file_name}
     * 版权：    Copyright by ljm
     * 描述：
     * 修改人：  HuamingChen
     * 修改时间：2018/10/23
     * 跟踪单号：
     * 修改单号：
     * 修改内容：
     */
    public class MQConsumer {
        private static final Logger logger= LoggerFactory.getLogger(MQConsumer.class);
        private static final String USERNAME= ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名
        private static final String PASSWORD= ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码
        private static final String BROKEURL= ActiveMQConnection.DEFAULT_BROKER_URL; // 默认的连接地址
    
        public static void main(String[] args){
            //连接工厂
            ConnectionFactory connectionFactory=null;
            //连接
            Connection connection=null;
            //会话
            Session session=null;
            //目的地
            Destination destination=null;
            //消费者
            MessageConsumer messageConsumer=null;
            //消息
            TextMessage textMessage=null;
    
            try {
                //创建连接工厂
                connectionFactory=new ActiveMQConnectionFactory(USERNAME,PASSWORD,BROKEURL);
                //创建连接
                connection=connectionFactory.createConnection();
                //启动连接
                connection.start();
                //创建会话
                session=connection.createSession(true,Session.AUTO_ACKNOWLEDGE);
                //创建目标
                //destination=session.createQueue("mingTest");
                destination=session.createTopic("mingTestTopic");
                //创建消费者
                messageConsumer=session.createConsumer(destination);
                //获取消息
                while (true){
                    textMessage= (TextMessage) messageConsumer.receive(10000);
                    if(textMessage!=null){
                        logger.info("收到的消息："+textMessage.getText());
                    }else{
                        break;
                    }
                }
                session.commit();
            } catch (JMSException e) {
                e.printStackTrace();
            } finally {
                if(connection!=null){
                    try {
                        connection.close();
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }

**2.4 消费者监听模式**

package com.ming.activitymq;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import javax.jms.*;
    
    /**
     * 文件名：  ${file_name}
     * 版权：    Copyright by ljm
     * 描述：
     * 修改人：  HuamingChen
     * 修改时间：2018/10/23
     * 跟踪单号：
     * 修改单号：
     * 修改内容：
     */
    public class MQConsumerListener {
        private static final Logger logger= LoggerFactory.getLogger(MQConsumerListener.class);
        private static final String USERNAME= ActiveMQConnection.DEFAULT_USER; // 默认的连接用户名
        private static final String PASSWORD= ActiveMQConnection.DEFAULT_PASSWORD; // 默认的连接密码
        private static final String BROKEURL= ActiveMQConnection.DEFAULT_BROKER_URL; // 默认的连接地址
    
        public static void main(String[] args){
            //连接工厂
            ConnectionFactory connectionFactory=null;
            //连接
            Connection connection=null;
            //会话
            Session session=null;
            //目的地
            Destination destination=null;
            //消费者
            MessageConsumer messageConsumer=null;
            //消息
            TextMessage textMessage=null;
    
            try {
                //创建连接工厂
                connectionFactory=new ActiveMQConnectionFactory(USERNAME,PASSWORD,BROKEURL);
                //创建连接
                connection=connectionFactory.createConnection();
                //启动连接
                connection.start();
                //创建会话
                session=connection.createSession(false,Session.AUTO_ACKNOWLEDGE);
                //创建目标
                //destination=session.createQueue("mingTest");//ptp模式
                destination=session.createTopic("mingTestTopic");//订阅模式
                //创建消费者
                messageConsumer=session.createConsumer(destination);
                //注册监听
                messageConsumer.setMessageListener(new MQListener());
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }

**2.5 实现监听接口**

package com.ming.activitymq;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import javax.jms.JMSException;
    import javax.jms.Message;
    import javax.jms.MessageListener;
    import javax.jms.TextMessage;
    
    /**
     * 文件名：  ${file_name}
     * 版权：    Copyright by ljm
     * 描述：    jms监听接口
     * 修改人：  HuamingChen
     * 修改时间：2018/10/23
     * 跟踪单号：
     * 修改单号：
     * 修改内容：
     */
    public class MQListener implements MessageListener {
        private static final Logger logger= LoggerFactory.getLogger(MQListener.class);
        @Override
        public void onMessage(Message message) {
            try {
                logger.info("收到的消息："+((TextMessage)message).getText());
            } catch (JMSException e) {
                e.printStackTrace();
            }
        }
    }

**完整代码：**  
[https://download.csdn.net/download/qq\_16055765/10739393][https_download.csdn.net_download_qq_16055765_10739393]

[https_download.csdn.net_download_qq_16055765_10739393]: https://download.csdn.net/download/qq_16055765/10739393