JMS操作ActiveMQ

叁歲伎倆 2022-02-04 11:49 428阅读 0赞

一般情况下我们都是通过jms操作mq，如果使用原生的jms代码比较繁琐，还好spring有集成的jms，通过jms模板类，能够适应各种mq的操作，其就像jdbc模板一样将具体的创建连接、连接的管理、消息的发送和接收等操作进行了封装，我们只需对mq连接池和jmsTemplate进行配置就可以了。

要使用mq进行消息发送和接收的缓存区，mq相当于是服务端，生产者和消费者都是客户端，生产者将消息发送到mq服务端，服务端再通知消费者消费，或是由消费者轮询mq服务端的队列进行消费，因此首先需要启动mq这个服务端。

一、启动ActiveMQ

这里以ActiveMQ为例，下一代ActiveMQ，Artemis的配置也类似。官网的操作指导地址：https://activemq.apache.org/getting-started

1、先从官网下载安装包：https://activemq.apache.org/index.html，这里以Windows为例进行安装，Linux操作相似。官网对于mq的安装环境有一定要求，比如jre版本。

2、解压缩安装包

3、进入命令目录bin

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5dXh1MzQyNzE5_size_16_color_FFFFFF_t_70][]

4、启动mq

不能直接双击activemq.bat，否则命令行窗口会闪退，可以通过cmd在命令行输入activemq start命令启动

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5dXh1MzQyNzE5_size_16_color_FFFFFF_t_70 1][]

使用netstat -an|find "61616"查看端口监听情况

![20190507172600300.png][]

如果想要修改监听端口可以修改conf下的activemq.xml文件

![20190507172600334.png][]

5、使用管理接口

打开地址 [http://127.0.0.1:8161/admin/][http_127.0.0.1_8161_admin]，登录用户和密码是admin，点击Queues菜单，查看注册到服务端的队列，且可以手动创建队列。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5dXh1MzQyNzE5_size_16_color_FFFFFF_t_70 2][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5dXh1MzQyNzE5_size_16_color_FFFFFF_t_70 3][]

二、spring集成jms操作mq

引入spring集成jms的包和activemq的包，其他spring包不特别列出，包括junit。

<dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-jms</artifactId>
        <version>4.3.20.RELEASE</version>
    </dependency>
    
    <dependency>
        <groupId>org.apache.activemq</groupId>
    	<artifactId>activemq-all</artifactId>
    	<version>5.15.9</version>
    </dependency>

1、首先定义连接工厂连接的服务端地址

<amq:connectionFactory id="connectionFactory" brokerURL="tcp://localhost:61616" >
    	</amq:connectionFactory>

通过引入activemq和jms的xml模式，可以通过标签元素简写bean的定义

xmlns:jms="http://www.springframework.org/schema/jms"
    xmlns:amq=http://activemq.apache.org/schema/core

2、定义mq缓存消息的队列

<amq:queue id="goodsQueue" physicalName="teriste.goodsQueue"></amq:queue>

3、定义操作连接工厂和队列的jmsTemplate类

4、引入消息生产者，其需要注入jmsTemplate进行消息发送操作，我们再代码中使用的是JmsOperations接口，其实现类就是JmsTemplate

![20190507172802786.png][]

5、然后是定义监听器容器，其中定义了监听队列的所有消费者

<jms:listener-container>
    		<jms:listener destination="teriste.goodsQueue" ref="consumer"
    		method="consumerMessage"/>
    	</jms:listener-container>

6、引入消费者

<?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://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-3.2.xsd
    		http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd
    		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
    	
    	
    	<amq:connectionFactory id="connectionFactory" brokerURL="tcp://localhost:61616" trustAllPackages="true">
    	</amq:connectionFactory>	
    	
    	<amq:queue id="goodsQueue" physicalName="teriste.goodsQueue"></amq:queue>
    	
    	<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
    		<constructor-arg name="connectionFactory" ref="connectionFactory" ></constructor-arg>
    		<property name="defaultDestinationName" value="teriste.goodsQueue" ></property>
    	</bean>
    	
    	<bean id="queryGoodsService" class="com.teriste.api.service.impl.QueryGoodsServiceImpl">
    		<constructor-arg name="jmsOperations" ref="jmsTemplate"></constructor-arg>
    	</bean>
    	
    	
    	<jms:listener-container>
    		<jms:listener destination="teriste.goodsQueue" ref="consumer"
    		method="consumerMessage"/>
    	</jms:listener-container>
    	
    	<bean id="consumer" class="com.teriste.api.service.impl.controller.MessageConsumerBusiness"/>
    </beans>

7、定义消息对象

public class ReqMessageDto implements Serializable {
        private static final long serialVersionUID = 388556503381616296L;
    
        private String name;
    
        private String storeCode;
    
        private String goodsCode;
    
        /**
         * @return the name
         */
        public String getName() {
            return name;
        }
    
        /**
         * @param name the name to set
         */
        public void setName(String name) {
            this.name = name;
        }
    
        /**
         * @return the storeCode
         */
        public String getStoreCode() {
            return storeCode;
        }
    
        /**
         * @param storeCode the storeCode to set
         */
        public void setStoreCode(String storeCode) {
            this.storeCode = storeCode;
        }
    
        /**
         * @return the goodsCode
         */
        public String getGoodsCode() {
            return goodsCode;
        }
    
        /**
         * @param goodsCode the goodsCode to set
         */
        public void setGoodsCode(String goodsCode) {
            this.goodsCode = goodsCode;
        }
    
        /**
         * @return the serialversionuid
         */
        public static long getSerialversionuid() {
            return serialVersionUID;
        }
    
        @Override
        public String toString() {
            return "ReqMessageDto [name=" + name + ", storeCode=" + storeCode + ", goodsCode=" + goodsCode + "]";
        }
    
    }

8、实现生产者

import org.springframework.jms.core.JmsOperations;
    import com.teriste.api.service.QueryGoodsService;
    import com.teriste.api.service.dto.request.ReqMessageDto;
    
    public class QueryGoodsServiceImpl implements QueryGoodsService {
        JmsOperations jmsOperations;
    
        public QueryGoodsServiceImpl(JmsOperations jmsOperations) {
            this.jmsOperations = jmsOperations;
        }
        @Override
        public void queryGoodsList() {
            ReqMessageDto reqMessageDto = new ReqMessageDto();
            reqMessageDto.setName("test");
            reqMessageDto.setGoodsCode("001001");
            reqMessageDto.setStoreCode("10002");
            jmsOperations.convertAndSend(reqMessageDto);
        }
    }

9、实现消费者

import com.teriste.api.service.dto.request.ReqMessageDto;
    
    public class MessageConsumerBusiness {
        public void consumerMessage(Object reqMessage) {
            if (reqMessage instanceof ReqMessageDto) {
                System.out.println((ReqMessageDto) reqMessage);
            }
        }
    }

10、测试类

import org.junit.Test;
    import org.springframework.context.support.ClassPathXmlApplicationContext;
    
    import com.teriste.api.service.QueryGoodsService;
    
    public class TestAmq {
    
        @Test
        public void sendMessage() {
            ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("conf/spring/spring-amq.xml");
            QueryGoodsService queryGoodsService = context.getBean(QueryGoodsService.class);
            queryGoodsService.queryGoodsList();
        }
    }

11、启动测试类

报如下错误，消息对象是不安全的

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5dXh1MzQyNzE5_size_16_color_FFFFFF_t_70 4][]

根据提示，打开地址http://activemq.apache.org/objectmessage.html

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5dXh1MzQyNzE5_size_16_color_FFFFFF_t_70 5][]

这里为了方便就设置所有包都是可信任的

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5dXh1MzQyNzE5_size_16_color_FFFFFF_t_70 6][]

<amq:connectionFactory id=*"connectionFactory"* brokerURL=*"tcp://localhost:61616"* trustAllPackages=*"true"*>

</amq:connectionFactory>

![2019050717315162.png][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5dXh1MzQyNzE5_size_16_color_FFFFFF_t_70]: /images/20220204/b6a7b632d3454802845b9c5df35f56e4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5dXh1MzQyNzE5_size_16_color_FFFFFF_t_70 1]: /images/20220204/30c936d240d64aa3b138b74eb02eadd2.png
[20190507172600300.png]: /images/20220204/c455363e5ffb46a48e42aeaf3afe4784.png
[20190507172600334.png]: /images/20220204/9634cf1a27ff40c598a559df95bea73a.png
[http_127.0.0.1_8161_admin]: http://127.0.0.1:8161/admin/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5dXh1MzQyNzE5_size_16_color_FFFFFF_t_70 2]: /images/20220204/bacd77337a5d4e94afe7cf0547fd6fed.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5dXh1MzQyNzE5_size_16_color_FFFFFF_t_70 3]: /images/20220204/3e877dc2cf0b482e9433212e9a8fd279.png
[20190507172802786.png]: /images/20220204/eea5f16f5d634c56826c02a38f283695.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5dXh1MzQyNzE5_size_16_color_FFFFFF_t_70 4]: /images/20220204/b4c1454c269b40c592456f2d561d22a3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5dXh1MzQyNzE5_size_16_color_FFFFFF_t_70 5]: /images/20220204/fbd2eb8fd88241d8b7048b5ea7c14d5d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2Rvbmd5dXh1MzQyNzE5_size_16_color_FFFFFF_t_70 6]: /images/20220204/0924cc6b78404deabbed9c657962bd7e.png
[2019050717315162.png]: /images/20220204/18c8e0112fd848d1a661049287baf391.png