Java message queue-detailed explanation of Spring integration with ActiveMq

This article mainly introduces the detailed explanation of Java message queue-Spring integration of ActiveMq. The editor thinks it is quite good, so I will share it with you now and give it as a reference. Let’s follow the editor to take a look

1. Overview

First of all, let’s review the Java message service with you. In the previous blog "Java Message Queue-JMS Overview", I analyzed for everyone:

1. Message service: a middleware used to solve the coupling between two or multiple programs , the bottom layer is implemented by Java.

2. Advantages: asynchronous, reliable

3. Message model: point-to-point, publish/subscribe

4.Objects in JMS

Then in another blog "Java Message Queue-ActiveMq in Practice", we started an ActiveMq project from 0 to 1. During the project development process, we had a certain understanding of ActiveMq. Understanding:

1. Write clients in multiple languages ​​and protocols. Languages: Java, C, C++, C#, Ruby, Perl, Python, PHP. Application protocol: OpenWire, Stomp REST, WS Notification, XMPP, AMQP

2. Fully supports JMS1.1 and J2EE 1.4 specifications (persistence, XA messages, transactions)

3. Support for Spring, ActiveMQ can be easily embedded into systems using Spring, and also supports the features of Spring 2.0

4. Through common J2EE servers (such as Geronimo , JBoss 4, GlassFish, WebLogic) testing, in which through the configuration of JCA 1.5 resource adapters, ActiveMQ can be automatically deployed to any commercial server compatible with J2EE 1.4

5. Support multiple transmissions Protocol: in-VM, TCP, SSL, NIO, UDP, JGroups, JXTA

6. Supports high-speed message persistence through JDBC and journal

7. The design ensures high-performance clustering, client-server, point-to-point

8. Supports Ajax

9. Supports integration with Axis

10. You can easily call the embedded JMS provider for testing

In the next blog, I will work with you to integrate Spring and ActiveMq. In this blog post, we are based on Spring+JMS+ActiveMQ+Tomcat implements Point-To-Point asynchronous queue messages and PUB/SUB (publish/subscribe) model. It is a simple example and does not include any business.

2. Directory structure

2.1 Project directory

IDE selected IDEA (recommended for everyone to use), in order to avoid downloading jar, the bottom layer uses maven to build a project, integrating Spring and ActiveMq

　

2.2 pom.xml

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://www.php.cn/">
 <modelVersion>4.0.0</modelVersion>
 <groupId>Crawl-Page</groupId>
 <artifactId>Crawl-Page</artifactId>
 <packaging>war</packaging>
 <version>1.0-SNAPSHOT</version>
 <name>Crawl-Page Maven Webapp</name>
 <url>http://www.php.cn/;/url>
 <!-- 版本管理 -->
 <properties>
 <springframework>4.1.8.RELEASE</springframework>
 </properties>

 <dependencies>
 <dependency>
  <groupId>junit</groupId>
  <artifactId>junit</artifactId>
  <version>4.10</version>
  <scope>test</scope>
 </dependency>


 <!-- JSP相关 -->
 <dependency>
 <groupId>jstl</groupId>
 <artifactId>jstl</artifactId>
 <version>1.2</version>
 </dependency>
 <dependency>
 <groupId>javax.servlet</groupId>
 <artifactId>servlet-api</artifactId>
 <scope>provided</scope>
 <version>2.5</version>
 </dependency>


 <!-- spring -->
 <dependency>
  <groupId>org.springframework</groupId>
  <artifactId>spring-core</artifactId>
  <version>${springframework}</version>
 </dependency>
 <dependency>
  <groupId>org.springframework</groupId>
  <artifactId>spring-context</artifactId>
  <version>${springframework}</version>
 </dependency>
 <dependency>
  <groupId>org.springframework</groupId>
  <artifactId>spring-tx</artifactId>
  <version>${springframework}</version>
 </dependency>
 <dependency>
  <groupId>org.springframework</groupId>
  <artifactId>spring-webmvc</artifactId>
  <version>${springframework}</version>
 </dependency>
 <dependency>
  <groupId>org.springframework</groupId>
  <artifactId>spring-jms</artifactId>
  <version>${springframework}</version>
 </dependency>
 <!-- xbean 如<amq:connectionFactory /> -->
 <dependency>
  <groupId>org.apache.xbean</groupId>
  <artifactId>xbean-spring</artifactId>
  <version>3.16</version>
 </dependency>

 <!-- activemq -->
 <dependency>
  <groupId>org.apache.activemq</groupId>
  <artifactId>activemq-core</artifactId>
  <version>5.7.0</version>
 </dependency>
 <dependency>
  <groupId>org.apache.activemq</groupId>
  <artifactId>activemq-pool</artifactId>
  <version>5.12.1</version>
 </dependency>

 <!-- 自用jar包，可以忽略-->
 <dependency>
  <groupId>commons-httpclient</groupId>
  <artifactId>commons-httpclient</artifactId>
  <version>3.1</version>
 </dependency>
 </dependencies>

 <build>
 <finalName>Crawl-Page</finalName>
 <plugins>
  <plugin>
  <groupId>org.apache.tomcat.maven</groupId>
  <artifactId>tomcat7-maven-plugin</artifactId>
  <configuration>
   <port>8080</port>
   <path>/</path>
  </configuration>
  </plugin>
 </plugins>
 </build>

</project>

Because the pom.xml file here is a bit long, I won’t expand it.

We can see that there are only a few dependencies, 1. Spring core dependency 2. ActiveMq core and pool (if students choose to import jar here, they can directly import the activemq mentioned in our previous blog -all this jar package) 3. Java servlet related dependencies

The dependency version of the ActiveMq pool we choose here will be related to the subsequent dtd, and the corresponding version is required, so students will wait until they configure the activemq file. , need to pay attention to the dtd version selection

2.3 web.xml

web.xml is also similar, specify the Spring configuration file, springMvc naming, encoding format

<?xml version="1.0" encoding="UTF-8"?>
<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://www.php.cn/"
   version="3.0">

 <display-name>Archetype Created Web Application</display-name>

 <!-- 加载spring的配置文件，例如hibernate、jms等集成 -->
 <context-param>
 <param-name>contextConfigLocation</param-name>
 <param-value>
  classpath:applicationContext*.xml;
 </param-value>
 </context-param>

 <listener>
 <listener-class>org.springframework.web.context.ContextLoaderListener</listener-class>
 </listener>

 <servlet>
 <servlet-name>springMVC</servlet-name>
 <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
 <init-param>
  <param-name>contextConfigLocation</param-name>
  <param-value>classpath:spring-mvc.xml</param-value>
 </init-param>
 <load-on-startup>1</load-on-startup>
 </servlet>
 <servlet-mapping>
 <servlet-name>springMVC</servlet-name>
 <url-pattern>/</url-pattern>
 </servlet-mapping>

 <!-- 处理编码格式 -->
 <filter>
 <filter-name>characterEncodingFilter</filter-name>
 <filter-class>org.springframework.web.filter.CharacterEncodingFilter</filter-class>
 <init-param>
  <param-name>encoding</param-name>
  <param-value>UTF-8</param-value>
 </init-param>
 <init-param>
  <param-name>forceEncoding</param-name>
  <param-value>true</param-value>
 </init-param>
 </filter>
 <filter-mapping>
 <filter-name>characterEncodingFilter</filter-name>
 <url-pattern>/*</url-pattern>
 </filter-mapping>

</web-app>

2.4 SpringMvc and applicationContext.xml

There is nothing special about SpringMVC. Students in need can refer to it:

<?xml version="1.0" encoding="UTF-8"?>
<!-- 查找最新的schemaLocation 访问 http://www.php.cn/ -->
<beans xmlns="http://www.springframework.org/schema/beans"
  xmlns:aop="http://www.springframework.org/schema/aop"
  xmlns:context="http://www.springframework.org/schema/context"
  xmlns:mvc="http://www.springframework.org/schema/mvc"
  xmlns:tx="http://www.springframework.org/schema/tx"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xsi:schemaLocation="http://www.springframework.org/schema/aop
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/">

 <!-- 启用MVC注解 -->
 <mvc:annotation-driven />
 <!-- 指定Sping组件扫描的基本包路径 -->
 <context:component-scan base-package="com.Jayce" >
  <!-- 这里只扫描Controller，不可重复加载Service -->
  <context:include-filter type="annotation" expression="org.springframework.stereotype.Controller"/>
 </context:component-scan>

 <!-- JSP视图解析器-->
 <bean class="org.springframework.web.servlet.view.InternalResourceViewResolver">
  <property name="prefix" value="/WEB-INF/views/" />
  <property name="suffix" value=".jsp" />
  <!-- 定义其解析视图的order顺序为1 -->
  <property name="order" value="1" />
 </bean>
</beans>

applicationContext.xml is mainly used to load beans. There are no special Java Beans in our project, so it is only used to indicate the package scanning path:

<?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"
  xmlns:context="http://www.springframework.org/schema/context"
  xmlns:mvc="http://www.springframework.org/schema/mvc"
  xsi:schemaLocation="
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/">

 <bean class="org.springframework.beans.factory.annotation.AutowiredAnnotationBeanPostProcessor"/>
 <!-- 配置扫描路径 -->
 <context:component-scan base-package="com.Jayce">
  <!-- 只扫描Service，也可以添加Repostory，但是要把Controller排除在外，Controller由spring-mvc.xml去加载 -->
  <context:exclude-filter type="annotation" expression="org.springframework.stereotype.Controller"/>
 </context:component-scan>

</beans>

2.5 applicationContext-ActiveMQ.xml

<?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"
  xmlns:context="http://www.springframework.org/schema/context"
  xmlns:mvc="http://www.springframework.org/schema/mvc"
  xsi:schemaLocation="
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/
  http://www.php.cn/"
>

 <context:component-scan base-package="com.Jayce" />
 <mvc:annotation-driven />

 <amq:connectionFactory id="amqConnectionFactory"
       brokerURL="tcp://192.168.148.128:61616"
       userName="admin"
       password="admin" />

 <!-- 配置JMS连接工长 -->
 <bean id="connectionFactory"
   class="org.springframework.jms.connection.CachingConnectionFactory">
  <constructor-arg ref="amqConnectionFactory" />
  <property name="sessionCacheSize" value="100" />
 </bean>

 <!-- 定义消息队列（Queue） -->
 <bean id="demoQueueDestination" class="org.apache.activemq.command.ActiveMQQueue">
  <!-- 设置消息队列的名字 -->
  <constructor-arg>
   <value>Jaycekon</value>
  </constructor-arg>
 </bean>

 <!-- 配置JMS模板（Queue），Spring提供的JMS工具类，它发送、接收消息。 -->
 <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
  <property name="connectionFactory" ref="connectionFactory" />
  <property name="defaultDestination" ref="demoQueueDestination" />
  <property name="receiveTimeout" value="10000" />
  <!-- true是topic，false是queue，默认是false，此处显示写出false -->
  <property name="pubSubDomain" value="false" />
 </bean>


 <!-- 配置消息队列监听者（Queue） -->
 <bean id="queueMessageListener" class="com.Jayce.Filter.QueueMessageListener" />

 <!-- 显示注入消息监听容器（Queue），配置连接工厂，监听的目标是demoQueueDestination，监听器是上面定义的监听器 -->
 <bean id="queueListenerContainer"
   class="org.springframework.jms.listener.DefaultMessageListenerContainer">
  <property name="connectionFactory" ref="connectionFactory" />
  <property name="destination" ref="demoQueueDestination" />
  <property name="messageListener" ref="queueMessageListener" />
 </bean>

</beans>

Here I will explain this configuration file to you. If you can see it from the above configuration file Got it, you can skip it. Students can also view it on the ActiveMQ official website.

1. DTD in ActiveMq, before we declare the relevant configuration, we need to import the DTD in ActiveMq, otherwise Spring will not understand what our tags mean.

We have configured the version of activemq in the pom.xml file that depends on our version here. It needs to be the same as the dependent version, otherwise the relevant dtd will not be found

2, amq: connectionFactory: A very straightforward configuration item, used to configure the address and username and password of our link factory. What needs to be noted here is to select the tcp connection instead of the http connection

3. jmsTemplate: a more important configuration , the connection factory, default message sending destination, connection duration, and message publishing method are specified here

3. Project structure

3.1 ProducerService

package com.Jayce.Service;

import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;

/**
 * Created by Administrator on 2017/1/5.
 */
@Service
public class ProducerService {

 @Resource(name="jmsTemplate")
 private JmsTemplate jmsTemplate;

 public void sendMessage(Destination destination,final String msg){
  System.out.println(Thread.currentThread().getName()+" 向队列"+destination.toString()+"发送消息---------------------->"+msg);
  jmsTemplate.send(destination, new MessageCreator() {
   public Message createMessage(Session session) throws JMSException {
    return session.createTextMessage(msg);
   }
  });
 }

 public void sendMessage(final String msg){
  String destination = jmsTemplate.getDefaultDestinationName();
  System.out.println(Thread.currentThread().getName()+" 向队列"+destination+"发送消息---------------------->"+msg);
  jmsTemplate.send(new MessageCreator() {
   public Message createMessage(Session session) throws JMSException {
    return session.createTextMessage(msg);
   }
  });
 }
}

Make the message producer into a service. When we need to send a message, we only need to call the sendMessage method in the ProducerService instance. You can send a message to the default destination.

Two sending methods are provided here, one is to send to the default destination, and the other is to send messages based on the destination.

3.2 ConsumerService

package com.Jayce.Service;

import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.TextMessage;

/**
 * Created by Administrator on 2017/1/5.
 */
@Service
public class ConsumerService {
 @Resource(name="jmsTemplate")
 private JmsTemplate jmsTemplate;

 public TextMessage receive(Destination destination){
  TextMessage textMessage = (TextMessage) jmsTemplate.receive(destination);
  try{
   System.out.println("从队列" + destination.toString() + "收到了消息：\t"
     + textMessage.getText());
  } catch (JMSException e) {
   e.printStackTrace();
  }
  return textMessage;
 }
}

因为我们项目中并没有什么业务，所以的话对消息的处理也就是打印输出。我们只需要调用jmsTemplate中的 receive 方法，就可以从里面获取到一条消息。

再和我们上一篇博客对比一下，上一篇博客中，我们接受到信息之后需要手动确认事务，这样ActiveMQ中才会确定这条消息已经被正确读取了。而整合了Spring之后，事务将由Spring 来管理。

3.3 MessageController

package com.Jayce.Controller;

import com.Jayce.Service.ConsumerService;
import com.Jayce.Service.ProducerService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;

import javax.annotation.Resource;
import javax.jms.Destination;
import javax.jms.TextMessage;

/**
 * Created by Administrator on 2017/1/5.
 */
@Controller
public class MessageController {
 private Logger logger = LoggerFactory.getLogger(MessageController.class);
 @Resource(name = "demoQueueDestination")
 private Destination destination;

 //队列消息生产者
 @Resource(name = "producerService")
 private ProducerService producer;

 //队列消息消费者
 @Resource(name = "consumerService")
 private ConsumerService consumer;

 @RequestMapping(value = "/SendMessage", method = RequestMethod.POST)
 @ResponseBody
 public void send(String msg) {
  logger.info(Thread.currentThread().getName()+"------------send to jms Start");
  producer.sendMessage(msg);
  logger.info(Thread.currentThread().getName()+"------------send to jms End");
 }

 @RequestMapping(value= "/ReceiveMessage",method = RequestMethod.GET)
 @ResponseBody
 public Object receive(){
  logger.info(Thread.currentThread().getName()+"------------receive from jms Start");
  TextMessage tm = consumer.receive(destination);
  logger.info(Thread.currentThread().getName()+"------------receive from jms End");
  return tm;
 }

}

控制层里面需要注入我们的生产者和消费者（实际开发中，生产者和消费者肯定不会在同一个项目中的，不然就消息服务这个东西就没有意义了）。

现在服务层和控制层都好了，接下来我们就进行一个简单的测试

 4、项目测试

4.1 启动ActiveMq

先确定你的ActiveMQ服务已经开启。

　　　　

4.2 启动项目

项目使用了Tomcat 插件，避免了本地再下载Tomcat的麻烦，有需要的同学可以使用一下。

<plugins>
  <plugin>
  <groupId>org.apache.tomcat.maven</groupId>
  <artifactId>tomcat7-maven-plugin</artifactId>
  <configuration>
   <port>8080</port>
   <path>/</path>
  </configuration>
  </plugin>
</plugins>

4.3 发送消息

这里用了Chrome 的一个插件PostMan 有兴趣的同学可以了解一下，在Chrome 拓展程序中可以找到，避免了后端的同学去弄页面！

　　　　

我们发送了一个post 请求之后，看一下服务器的效果：

 

我们可以看到，已经向队列发送了一条消息。我们看一下ActiveMq现在的状态：

 

我们可以看到，一条消息已经成功发送到了ActiveMq中。

4.4 接收消息

使用get请求访问服务器后台：

　　

服务的输出：

 

ActiveMq服务器状态：

 

我们可以看到，消费者已经消费了一条信息，并且没有断开与ActiveMq之间的链接。

4.5 监听器

在实际项目中，我们很少会自己手动去获取消息，如果需要手动去获取消息，那就没有必要使用到ActiveMq了，可以用一个Redis 就足够了。

不能手动去获取消息，那么我们就可以选择使用一个监听器来监听是否有消息到达，这样子可以很快的完成对消息的处理。

4.5.1 applicationContext-ActiveMQ.xml 配置

在上面的配置文件中，我们已经默认的添加了这段监听器的配置文件，如果同学们不想使用这个监听器，可以直接注释掉。

  

 <!-- 配置消息队列监听者（Queue） -->
 <bean id="queueMessageListener" class="com.Jayce.Filter.QueueMessageListener" />

 <!-- 显示注入消息监听容器（Queue），配置连接工厂，监听的目标是demoQueueDestination，监听器是上面定义的监听器 -->
 <bean id="queueListenerContainer"
   class="org.springframework.jms.listener.DefaultMessageListenerContainer">
  <property name="connectionFactory" ref="connectionFactory" />
  <property name="destination" ref="demoQueueDestination" />
  <property name="messageListener" ref="queueMessageListener" />
 </bean>

4.5.2 MessageListener

我们需要创建一个类实现MessageListener 接口：

package com.Jayce.Filter;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;

/**
 * Created by Administrator on 2017/1/5.
 */
public class QueueMessageListener implements MessageListener {
 public void onMessage(Message message) {
  TextMessage tm = (TextMessage) message;
  try {
   System.out.println("QueueMessageListener监听到了文本消息：\t"
     + tm.getText());
   //do something ...
  } catch (JMSException e) {
   e.printStackTrace();
  }
 }
}

实现接口的onMessage 方法，我们将需要的业务操作在里面解决，这样子，就完成了我们生产者-中间件-消费者，这样一个解耦的操作了。　

4.5.3 测试

和上面一样，使用postMan 发送post请求，我们可以看到控制台里面，消息马上就能打印出来：

　　　　

再看看ActiveMQ服务器的状态：

 

我们可以看到，使用监听器的效果，和手动接收消息的效果是一样的。

这样子一整个项目下来，我们已经成功的整合了Spring和ActiveMQ。

4.6 压力测试

这里其实也算不上什么压力测试，在配置pom.xml文件的时候，大家有看到一个 commons-httpclient 的依赖，接下来我们使用httpClient 不停的想服务器发送消息，看一下服务器解决消息的速度如何：

package com.Jaycekon.test;

import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.methods.PostMethod;
import org.junit.Test;

import java.io.IOException;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by Administrator on 2017/1/5.
 */
public class Client {

 @Test
 public void test() {
  HttpClient httpClient = new HttpClient();
  new Thread(new Sender(httpClient)).start();

 }

}

class Sender implements Runnable {
 public static AtomicInteger count = new AtomicInteger(0);
 HttpClient httpClient;

 public Sender(HttpClient client) {
  httpClient = client;
 }

 public void run() {
   try {
    System.out.println(Thread.currentThread().getName()+"---Send message-"+count.getAndIncrement());
    PostMethod post = new PostMethod("http://127.0.0.1:8080/SendMessage");
    post.addParameter("msg", "Hello world!");
    httpClient.executeMethod(post);
    System.out.println(Thread.currentThread().getName()+"---Send message Success-"+count.getAndIncrement());

   } catch (IOException e) {
    e.printStackTrace();
   }
  }
 }

这里面用了HttpClient 来向服务器发送Post 请求，然后计数输出，有兴趣的同学可以自己测试一下，可以多开几个线程，这里只开了一个线程。

5、项目源码：Crawl-Page_jb51.rar

The above is the detailed explanation of Java Message Queue-Spring integrating ActiveMq. For more related content, please pay attention to the PHP Chinese website (www.php.cn)!