jms基础入门(activemq)

秒速五厘米 2022-06-03 09:00 290阅读 0赞

ActiveMQ 是Apache出品，最流行的，能力强劲的开源消息总线。ActiveMQ 是一个完全支持JMS1.1和J2EE 1.4规范的 JMS Provider实现，尽管JMS规范出台已经是很久的事情了，但是JMS在当今的J2EE应用中间仍然扮演着特殊的地位。

1.ActiveMQ安装与使用

点我下载

下载后,进入目录apache-activemq-5.14.0\bin\win32|64
这里写图片描述

当然,还可以通过命令开启,

  • 普通启动 ./activemq start
  • 启动并指定日志文件 ./activemq start >tmp/smlog
  • 后台启动方式nohup ./activemq start >/tmp/smlog

运行成功后,打开网页,访问http://localhost:8161/admin/
第一次登陆,会要求输入密码 , 默认账号admin 默认密码admin

这里写图片描述
以上是登陆成功的页面,不同版本可能页面不同,但功能能用就证明登陆成功,

java程序演示

创建一个java项目,导入jar包,
这里写图片描述

  1. 创建一个会话创建者,
  2. package cn.itcast.demo;
  4. import javax.jms.Connection;
  5. import javax.jms.ConnectionFactory;
  6. import javax.jms.Destination;
  7. import javax.jms.JMSException;
  8. import javax.jms.MessageProducer;
  9. import javax.jms.Session;
  10. import javax.jms.TextMessage;
  12. import org.apache.activemq.ActiveMQConnection;
  13. import org.apache.activemq.ActiveMQConnectionFactory;
  15. public class JMSProducer { 
  16.     // 默认连接用户名
  17.     private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
  18.     // 默认连接密码
  19.     private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
  20.     // 默认连接地址
  21.     private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;
  22.     // 发送的消息数量
  23.     private static final int SENDNUM = 10;
  25.     public static void main(String[] args) {
  26.         // 连接工厂
  27.         ConnectionFactory connectionFactory;
  28.         // 连接
  29.         Connection connection = null;
  30.         // 会话 接受或者发送消息的线程
  31.         Session session;
  32.         // 消息的目的地
  33.         Destination destination;
  34.         // 消息生产者
  35.         MessageProducer messageProducer;
  36.         // 实例化连接工厂
  37.         connectionFactory = new ActiveMQConnectionFactory(JMSProducer.USERNAME, JMSProducer.PASSWORD,
  38.                 JMSProducer.BROKEURL);
  40.         try {
  41.             // 通过连接工厂获取连接
  42.             connection = connectionFactory.createConnection();
  43.             // 启动连接
  44.             connection.start();
  45.             // 创建session
  46.             session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
  47.             // 创建一个名称为HelloWorld的消息队列
  48.             destination = session.createQueue("HelloWorld");
  49.             // 创建消息生产者
  50.             messageProducer = session.createProducer(destination);
  51.             // 发送消息
  52.             sendMessage(session, messageProducer);
  54.             session.commit();
  56.         } catch (Exception e) {
  57.             e.printStackTrace();
  58.         } finally {
  59.             if (connection != null) {
  60.                 try {
  61.                     connection.close();
  62.                 } catch (JMSException e) {
  63.                     e.printStackTrace();
  64.                 }
  65.             }
  66.         }
  68.     }
  70.     /** * 发送消息 * * @param session * @param messageProducer * 消息生产者 * @throws Exception */
  71.     public static void sendMessage(Session session, MessageProducer messageProducer) throws Exception {
  72.         for (int i = 0; i < JMSProducer.SENDNUM; i++) {
  73.             // 创建一条文本消息
  74.             TextMessage message = session.createTextMessage("ActiveMQ 发送消息" + i);
  75.             System.out.println("发送消息：Activemq 发送消息" + i);
  76.             // 通过消息生产者发出消息
  77.             messageProducer.send(message);
  78.         }
  80.     }
  81. }

运行如上代码测试
这里写图片描述

然后我们进入管理页面看看
这里写图片描述

我们创建一个消费者,

  1. package cn.itcast.demo;
  3. import javax.jms.Connection;
  4. import javax.jms.ConnectionFactory;
  5. import javax.jms.Destination;
  6. import javax.jms.JMSException;
  7. import javax.jms.MessageConsumer;
  8. import javax.jms.Session;
  9. import javax.jms.TextMessage;
  11. import org.apache.activemq.ActiveMQConnection;
  12. import org.apache.activemq.ActiveMQConnectionFactory;
  14. public class JMSConsumer { 
  15.     private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;// 默认连接用户名
  16.     private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;// 默认连接密码
  17.     private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;// 默认连接地址
  19.     public static void main(String[] args) {
  20.         ConnectionFactory connectionFactory;// 连接工厂
  21.         Connection connection = null;// 连接
  23.         Session session;// 会话 接受或者发送消息的线程
  24.         Destination destination;// 消息的目的地
  26.         MessageConsumer messageConsumer;// 消息的消费者
  28.         // 实例化连接工厂
  29.         connectionFactory = new ActiveMQConnectionFactory(JMSConsumer.USERNAME, JMSConsumer.PASSWORD,
  30.                 JMSConsumer.BROKEURL);
  32.         try {
  33.             // 通过连接工厂获取连接
  34.             connection = connectionFactory.createConnection();
  35.             // 启动连接
  36.             connection.start();
  37.             // 创建session
  38.             session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
  39.             // 创建一个连接HelloWorld的消息队列
  40.             destination = session.createQueue("HelloWorld");
  41.             // 创建消息消费者
  42.             messageConsumer = session.createConsumer(destination);
  44.             while (true) {
  45.                 TextMessage textMessage = (TextMessage) messageConsumer.receive(100000);
  46.                 if (textMessage != null) {
  47.                     System.out.println("收到的消息:" + textMessage.getText());
  48.                 } else {
  49.                     break;
  50.                 }
  51.             }
  53.         } catch (JMSException e) {
  54.             e.printStackTrace();
  55.         }
  57.     }
  58. }

运行结果如下
这里写图片描述

然后我们再进入到管理页面
这里写图片描述

其实这些代码没有什么技术量,,复制粘贴就行,,唯一需要改的就是消息队列名了,注意一下就行,

maven整合spring+jms(activemq)

applicationContext.xml(模板如下)

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:aop="http://www.springframework.org/schema/aop" xmlns:context="http://www.springframework.org/schema/context" xmlns:jdbc="http://www.springframework.org/schema/jdbc" xmlns:tx="http://www.springframework.org/schema/tx" xmlns:jpa="http://www.springframework.org/schema/data/jpa" xmlns:task="http://www.springframework.org/schema/task" 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.1.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.1.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.1.xsd http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc-4.1.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.1.xsd http://www.springframework.org/schema/data/jpa http://www.springframework.org/schema/data/jpa/spring-jpa.xsd http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms.xsd http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.8.0.xsd ">
  4.     <!-- 扫盲包 -->
  5.     <context:component-scan base-package="cn.itcast.activemq" />
  7.     <!-- ActiveMQ 连接工厂 -->
  8.     <!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供-->
  9.     <!-- 如果连接网络：tcp://ip:61616；未连接网络：tcp://localhost:61616 以及用户名，密码-->
  10.     <amq:connectionFactory id="amqConnectionFactory" brokerURL="tcp://localhost:61616" userName="admin" password="admin" />
  12.     <!-- Spring Caching连接工厂 -->
  13.     <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->  
  14.     <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
  15.         <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->  
  16.         <property name="targetConnectionFactory" ref="amqConnectionFactory"></property>
  17.         <!-- 同上，同理 -->
  18.         <!-- <constructor-arg ref="amqConnectionFactory" /> -->
  19.         <!-- Session缓存数量 -->
  20.         <property name="sessionCacheSize" value="100" />
  21.     </bean>
  23.      <!-- Spring JmsTemplate 的消息生产者 start-->
  25.     <!-- 定义JmsTemplate的Queue类型 -->
  26.     <bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">
  27.         <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->  
  28.         <constructor-arg ref="connectionFactory" />
  29.         <!-- 非pub/sub模型（发布/订阅），即队列模式 -->
  30.         <property name="pubSubDomain" value="false" />
  31.     </bean>
  33.         这也是二选一
  35.     <!-- 定义JmsTemplate的Topic类型 -->
  36.     <bean id="jmsTopicTemplate" class="org.springframework.jms.core.JmsTemplate">
  37.          <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->  
  38.         <constructor-arg ref="connectionFactory" />
  39.         <!-- pub/sub模型（发布/订阅） -->
  40.         <property name="pubSubDomain" value="true" />
  41.     </bean>
  43.     <!--Spring JmsTemplate 的消息生产者 end-->
  45.     <!-- 消息消费者 start-->
  47.     <!-- 定义Queue监听器 -->
  48.     <jms:listener-container destination-type="queue" container-type="default" connection-factory="connectionFactory" acknowledge="auto">
  49.      <!--消息队列名称 消费者类名,首字母小写,需要加上注解才行 -->
  50.         <jms:listener destination="test.queue" ref="queueReceiver1"/>
  51.     </jms:listener-container>
  55.     俩种监听方式,一个是一对一,一个是一对多,区别我就不多说了,不会自行百度,通常开发都是二选一,自己选择合适的
  57.     <!-- 定义Topic监听器 -->
  58.     <jms:listener-container destination-type="topic" container-type="default" connection-factory="connectionFactory" acknowledge="auto">
  59.         <jms:listener destination="test.topic" ref="topicReceiver1"/>
  60.         <jms:listener destination="test.topic" ref="topicReceiver2"/>
  61.     </jms:listener-container>
  63.     <!-- 消息消费者 end -->
  66. </beans>

pom依赖

  1. <dependencies>
  2.     <dependency>
  3.         <groupId>junit</groupId>
  4.         <artifactId>junit</artifactId>
  5.         <version>4.12</version>
  6.     </dependency>
  7.     <dependency>
  8.         <groupId>org.springframework</groupId>
  9.         <artifactId>spring-core</artifactId>
  10.         <version>4.1.7.RELEASE</version>
  11.     </dependency>
  12.     <dependency>
  13.         <groupId>org.springframework</groupId>
  14.         <artifactId>spring-jms</artifactId>
  15.         <version>4.1.7.RELEASE</version>
  16.     </dependency>
  17.     <dependency>
  18.         <groupId>org.apache.activemq</groupId>
  19.         <artifactId>activemq-all</artifactId>
  20.         <version>5.2.0</version>
  21.     </dependency>
  22.     <dependency>
  23.         <groupId>org.springframework</groupId>
  24.         <artifactId>spring-test</artifactId>
  25.         <version>4.1.7.RELEASE</version>
  26.     </dependency>
  28.     <!-- <dependency> <groupId>org.apache.xbean</groupId> <artifactId>xbean-spring</artifactId> <version>4.2</version> </dependency> -->
  29.   </dependencies>

好了,我们开始撸代码(做一个注册发短信提示激活功能)

首先写消费者

0.创建一个maven工程

1.复制原先准备好的pom文件

2.编写web.xml

  1. <web-app xmlns="http://java.sun.com/xml/ns/javaee" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd" version="2.5">
  3.     <listener>
  4.         <listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>
  5.     </listener>
  7.     <!-- 指定spring配置文件的位置,文件放在resources下 -->
  8.     <context-param>
  9.         <param-name>contextConfigLocation</param-name>
  10.         <param-value>classpath:applicationContext.xml</param-value>
  11.     </context-param>
  12. </web-app>

3.配置applicationContext.xml文件(放在resources目录下)

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:aop="http://www.springframework.org/schema/aop" xmlns:context="http://www.springframework.org/schema/context" xmlns:jdbc="http://www.springframework.org/schema/jdbc" xmlns:tx="http://www.springframework.org/schema/tx" xmlns:jpa="http://www.springframework.org/schema/data/jpa" xmlns:task="http://www.springframework.org/schema/task" 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.1.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-4.1.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-4.1.xsd http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc-4.1.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-4.1.xsd http://www.springframework.org/schema/data/jpa http://www.springframework.org/schema/data/jpa/spring-jpa.xsd http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms.xsd http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.8.0.xsd ">
  4.     <!-- 扫盲包 -->
  5.     <context:component-scan base-package="cn.itcast.activemq" />
  7.     <!-- ActiveMQ 连接工厂 -->
  8.     <!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供-->
  9.     <!-- 如果连接网络：tcp://ip:61616；未连接网络：tcp://localhost:61616 以及用户名，密码-->
  10.     <amq:connectionFactory id="amqConnectionFactory" brokerURL="tcp://localhost:61616" userName="admin" password="admin" />
  12.     <!-- Spring Caching连接工厂 -->
  13.     <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->  
  14.     <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
  15.         <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->  
  16.         <property name="targetConnectionFactory" ref="amqConnectionFactory"></property>
  17.         <!-- 同上，同理 -->
  18.         <!-- <constructor-arg ref="amqConnectionFactory" /> -->
  19.         <!-- Session缓存数量 -->
  20.         <property name="sessionCacheSize" value="100" />
  21.     </bean>
  23.     <!-- 消息消费者 start-->
  25.     <!-- 定义Queue监听器 -->
  26.     <jms:listener-container destination-type="queue" container-type="default" connection-factory="connectionFactory" acknowledge="auto">
  28.         <jms:listener destination="sms.queue" ref="registerQueue"/>
  29.     </jms:listener-container>
  30.     <!-- 消息消费者 end -->
  32. </beans>

编写消费者类

需要给applicationContext的包扫描扫到,

  1. package cn.itcast.activemq;
  3. import org.slf4j.Logger;
  4. import org.slf4j.LoggerFactory;
  5. import org.springframework.stereotype.Component;
  7. import javax.jms.JMSException;
  8. import javax.jms.MapMessage;
  9. import javax.jms.Message;
  10. import javax.jms.MessageListener;
  12. /** * 发送短信系统, */
  13. @Component
  14. public class RegisterQueue implements MessageListener { 
  15.     private Logger log = LoggerFactory.getLogger(RegisterQueue.class);
  16.     public void onMessage(Message message) {
  18.         //接收map类型数据
  19.         MapMessage map = (MapMessage)message;
  20.         try {
  21.             String telephone = map.getString("telephone"); //接收电话
  22.             String content = map.getString("content") ;  //这是生产者存入的字段名
  23.             log.info("telephone:"+telephone);
  24.             log.info("content:"+content);
  25.             log.info("模拟发送短信");
  26.         } catch (JMSException e) {
  27.             e.printStackTrace();
  28.         }
  29.     }
  30. }

最后我们需要注意的是,因为是一个单独的系统,所以我们需要配置一个tomcat,在pom文件里面

  1. <plugins>
  2.       <plugin>
  3.         <groupId>org.apache.tomcat.maven</groupId>
  4.         <artifactId>tomcat7-maven-plugin</artifactId>
  5.         <version>2.2</version>
  6.         <configuration>
  7.           <uriEncoding>UTF-8</uriEncoding>
  8.           <path>/bos_sms</path>
  9.           <port>8892</port>
  10.         </configuration>
  11.       </plugin>
  12.     </plugins>

消费者已经写完了,然后我们开始写生产者(客户端),

由于客户端涉及相关业务,不能发完整代码,体谅一下

1.创建一个maven工程,

2.配置web.xml文件,

  1. <-- 配置web.xml -->
  2. <!--struts2拦截器-->
  3. <filter>
  4.         <filter-name>struts2</filter-name>
  5.         <filter-class>org.apache.struts2.dispatcher.ng.filter.StrutsPrepareAndExecuteFilter</filter-class>
  6.     </filter>
  7.     <filter-mapping>
  8.         <filter-name>struts2</filter-name>
  9.         <url-pattern>*</url-pattern>
  10.     </filter-mapping>
  12. <!--spring初始化配置-->
  13. <context-param>
  14.         <param-name>contextConfigLocation</param-name>
  15.         <param-value>classpath:applicationContext.xml</param-value>
  16.     </context-param>
  18.     <!-- 配置spring框架的监听器 -->
  19.     <listener>
  20.         <listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>
  21.     </listener>

配置applicationContext.xml

  1. <!--把这下面的东西拷贝进去修改修改就行了,-->
  3.     <!-- 扫盲包 -->
  4.     <context:component-scan base-package="cn.itcast.activemq" />
  6.     <!-- ActiveMQ 连接工厂 -->
  7.     <!-- 真正可以产生Connection的ConnectionFactory，由对应的 JMS服务厂商提供-->
  8.     <!-- 如果连接网络：tcp://ip:61616；未连接网络：tcp://localhost:61616 以及用户名，密码-->
  9.     <amq:connectionFactory id="amqConnectionFactory" brokerURL="tcp://localhost:61616" userName="admin" password="admin" />
  11.     <!-- Spring Caching连接工厂 -->
  12.     <!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->  
  13.     <bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
  14.         <!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->  
  15.         <property name="targetConnectionFactory" ref="amqConnectionFactory"></property>
  16.         <!-- 同上，同理 -->
  17.         <!-- <constructor-arg ref="amqConnectionFactory" /> -->
  18.         <!-- Session缓存数量 -->
  19.         <property name="sessionCacheSize" value="100" />
  20.     </bean>
  22.      <!-- Spring JmsTemplate 的消息生产者 start-->
  24.     <!-- 定义JmsTemplate的Queue类型 -->
  25.     <bean id="jmsQueueTemplate" class="org.springframework.jms.core.JmsTemplate">
  26.         <!-- 这个connectionFactory对应的是我们定义的Spring提供的那个ConnectionFactory对象 -->  
  27.         <constructor-arg ref="connectionFactory" />
  28.         <!-- 非pub/sub模型（发布/订阅），即队列模式 -->
  29.         <property name="pubSubDomain" value="false" />
  30.     </bean>

创建一个生产者

  1. @Resource
  2.     private JmsTemplate template;
  5.    @Action(value = "customerAction_regist",results = {
  6.    @Result(name = "fail",location = "/signup-fail.html",type = "redirect"),
  7.            @Result(name = "success",location = "/signup-success.html",type = "redirect")})
  8.    public String regist(){
  10.        String sms_name = "sms.queue";//消息队列明,需要跟我们先前设置消费者的一样
  11.        final String content = "恭喜你,你的账号已注册,请登陆邮箱激活!";//消息内容
  13.        //创建一个消息匿名类
  14.        template.send(sms_name, new MessageCreator() {
  15.            public Message createMessage(Session session) throws JMSException {
  16.                MapMessage mapMessage = session.createMapMessage();//获取map的消息对象
  17.                mapMessage.setString("telephone",customer.getTelephone());
  18.                mapMessage.setString("content",content);
  19.                return mapMessage;
  20.            }
  21.        });

最后还需要配置一个tomcat ,

  1. <plugins>
  2.       <plugin>
  3.         <groupId>org.apache.tomcat.maven</groupId>
  4.         <artifactId>tomcat7-maven-plugin</artifactId>
  5.         <version>2.2</version>
  6.         <configuration>
  7.           <uriEncoding>UTF-8</uriEncoding>
  8.           <path>/bos_before</path>
  9.           <port>8890</port>
  10.         </configuration>
  11.       </plugin>
  12.     </plugins>

最后启动运行试试,需要注意的是,俩个项目都要启动,
这里写图片描述

sms服务启动成功,接下来我们启动另外一个服务,
这里写图片描述

也启动成功,我们开始代码测试

测试抛出异常

  1. Exception occurred during processing request: null
  2. java.lang.NullPointerException

根据提示信息发现 , JmsTemplate template;注入失败,

然后一顿乱找,发现是切面的影响的,,这个有点坑

把这行代码注释掉就没问题了

  1. <!--使用切面注解扫描-->
  2.     <aop:aspectj-autoproxy proxy-target-class="true"/>

原因是代理类不能使用JmsTemplate,

最后运行是没问题的,

这里写图片描述

jms就讲到这里,提前祝大家圣诞节快乐

发表评论

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

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

相关阅读

    相关 activeMQ Jms Demo

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

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

    相关 JMS操作ActiveMQ

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

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