JMS操作ActiveMQ

叁歲伎倆 2022-02-04 11:49 484阅读 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/，登录用户和密码是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。

  1. <!-- https://mvnrepository.com/artifact/org.springframework/spring-jms -->
  2. <dependency>
  3.     <groupId>org.springframework</groupId>
  4.     <artifactId>spring-jms</artifactId>
  5.     <version>4.3.20.RELEASE</version>
  6. </dependency>
  7. <!-- https://mvnrepository.com/artifact/org.apache.activemq/activemq-all -->
  8. <dependency>
  9.     <groupId>org.apache.activemq</groupId>
  10.     <artifactId>activemq-all</artifactId>
  11.     <version>5.15.9</version>
  12. </dependency>

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

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

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

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

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

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

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

  1. <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
  2.         <constructor-arg name="connectionFactory" ref="connectionFactory" ></constructor-arg>
  3.         <property name="defaultDestinationName" value="teriste.goodsQueue" ></property>
  4.     </bean>

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

  1. <bean id="queryGoodsService" class="com.teriste.api.service.impl.QueryGoodsServiceImpl">
  2.         <constructor-arg name="jmsOperations" ref="jmsTemplate"></constructor-arg>
  3.     </bean>

20190507172802786.png

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

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

6、引入消费者

  1. <bean id="consumer" class="com.teriste.api.controller.MessageConsumerBusiness"/>
  3. <?xml version="1.0" encoding="UTF-8"?>
  4. <beans xmlns="http://www.springframework.org/schema/beans"
  5.     xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  6.     xmlns:jms="http://www.springframework.org/schema/jms"
  7.     xmlns:amq="http://activemq.apache.org/schema/core"
  8.     xsi:schemaLocation="http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-3.2.xsd
  9.         http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd
  10.         http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
  11.     <!-- 创建连接池连接amq，url是再activemq.xml中配置的地址 -->
  12.     <!-- 有了amq:connectionFactory注解就不需要显示定义class属性
  13.      org.apache.activemq.spring.ActiveMQConnectionFactory-->
  14.     <amq:connectionFactory id="connectionFactory" brokerURL="tcp://localhost:61616" trustAllPackages="true">
  15.     </amq:connectionFactory>    
  16.     <!-- 配置监听队列监听生产者发送的消息 -->
  17.     <amq:queue id="goodsQueue" physicalName="teriste.goodsQueue"></amq:queue>
  18.     <!-- jms模板，用于建立连接和操作队列 -->
  19.     <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
  20.         <constructor-arg name="connectionFactory" ref="connectionFactory" ></constructor-arg>
  21.         <property name="defaultDestinationName" value="teriste.goodsQueue" ></property>
  22.     </bean>
  23.     <!-- 生产者通过jmsOperations向队列中发送消息，其实现类就是 jmsTemplate-->
  24.     <bean id="queryGoodsService" class="com.teriste.api.service.impl.QueryGoodsServiceImpl">
  25.         <constructor-arg name="jmsOperations" ref="jmsTemplate"></constructor-arg>
  26.     </bean>
  28.     <!-- jms监听器，即消费者监听队列产生的消息进行消费 -->
  29.     <jms:listener-container>
  30.         <jms:listener destination="teriste.goodsQueue" ref="consumer"
  31.         method="consumerMessage"/>
  32.     </jms:listener-container>
  33.     <!-- 消费者 -->
  34.     <bean id="consumer" class="com.teriste.api.service.impl.controller.MessageConsumerBusiness"/>
  35. </beans>

7、定义消息对象

  1. public class ReqMessageDto implements Serializable {
  2.     private static final long serialVersionUID = 388556503381616296L;
  4.     private String name;
  6.     private String storeCode;
  8.     private String goodsCode;
  10.     /**
  11.      * @return the name
  12.      */
  13.     public String getName() {
  14.         return name;
  15.     }
  17.     /**
  18.      * @param name the name to set
  19.      */
  20.     public void setName(String name) {
  21.         this.name = name;
  22.     }
  24.     /**
  25.      * @return the storeCode
  26.      */
  27.     public String getStoreCode() {
  28.         return storeCode;
  29.     }
  31.     /**
  32.      * @param storeCode the storeCode to set
  33.      */
  34.     public void setStoreCode(String storeCode) {
  35.         this.storeCode = storeCode;
  36.     }
  38.     /**
  39.      * @return the goodsCode
  40.      */
  41.     public String getGoodsCode() {
  42.         return goodsCode;
  43.     }
  45.     /**
  46.      * @param goodsCode the goodsCode to set
  47.      */
  48.     public void setGoodsCode(String goodsCode) {
  49.         this.goodsCode = goodsCode;
  50.     }
  52.     /**
  53.      * @return the serialversionuid
  54.      */
  55.     public static long getSerialversionuid() {
  56.         return serialVersionUID;
  57.     }
  59.     @Override
  60.     public String toString() {
  61.         return "ReqMessageDto [name=" + name + ", storeCode=" + storeCode + ", goodsCode=" + goodsCode + "]";
  62.     }
  64. }

8、实现生产者

  1. import org.springframework.jms.core.JmsOperations;
  2. import com.teriste.api.service.QueryGoodsService;
  3. import com.teriste.api.service.dto.request.ReqMessageDto;
  5. public class QueryGoodsServiceImpl implements QueryGoodsService {
  6.     JmsOperations jmsOperations;
  8.     public QueryGoodsServiceImpl(JmsOperations jmsOperations) {
  9.         this.jmsOperations = jmsOperations;
  10.     }
  11.     @Override
  12.     public void queryGoodsList() {
  13.         ReqMessageDto reqMessageDto = new ReqMessageDto();
  14.         reqMessageDto.setName("test");
  15.         reqMessageDto.setGoodsCode("001001");
  16.         reqMessageDto.setStoreCode("10002");
  17.         jmsOperations.convertAndSend(reqMessageDto);
  18.     }
  19. }

9、实现消费者

  1. import com.teriste.api.service.dto.request.ReqMessageDto;
  3. public class MessageConsumerBusiness {
  4.     public void consumerMessage(Object reqMessage) {
  5.         if (reqMessage instanceof ReqMessageDto) {
  6.             System.out.println((ReqMessageDto) reqMessage);
  7.         }
  8.     }
  9. }

10、测试类

  1. import org.junit.Test;
  2. import org.springframework.context.support.ClassPathXmlApplicationContext;
  4. import com.teriste.api.service.QueryGoodsService;
  6. public class TestAmq {
  8.     @Test
  9.     public void sendMessage() {
  10.         ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("conf/spring/spring-amq.xml");
  11.         QueryGoodsService queryGoodsService = context.getBean(QueryGoodsService.class);
  12.         queryGoodsService.queryGoodsList();
  13.     }
  14. }

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

2019050717315162.png

发表评论

表情：
评论列表 （有 0 条评论，484人围观）

还没有评论，来说两句吧...

相关阅读

    相关 JMS-ActiveMQ系列

    个人总结:消息队列，ActiveMQ现在的企业用得很广泛，分点对点（生产消费），发布/订阅两种。现开发的爬虫系统，多个worker抓取到的信息发到mq，mq监听器把信息异步存到

    分手后的思念是犯贱分手后的思念是犯贱/ 0 赞/ 315 阅读

    相关 activeMQ Jms Demo

    概述 ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provide

    女爷i女爷i/ 0 赞/ 358 阅读

    相关 JMS操作ActiveMQ

    一般情况下我们都是通过jms操作mq，如果使用原生的jms代码比较繁琐，还好spring有集成的jms，通过jms模板类，能够适应各种mq的操作，其就像jdbc模板一样将具体的

    叁歲伎倆叁歲伎倆/ 0 赞/ 485 阅读