Java Message Queue - Spring の ActiveMq 統合の詳細な説明

この記事では、Java メッセージ キューと Spring 統合 ActiveMq の詳細な説明を主に紹介します。編集者が非常に優れていると考えたので、参考として共有します。エディターと一緒に見てみましょう

1. 概要

まず最初に、以前のブログ「Java Message Queue-JMS の概要」で Java メッセージ サービスを分析しました。 for you :

1. メッセージ サービス: 2 つまたは複数のプログラム間の結合を解決するために使用されるミドルウェア。最下層は Java によって実装されます。

2. 利点: 非同期、信頼性

3. メッセージ モデル: ポイントツーポイント、パブリッシュ/サブスクライブ

4. JMS のオブジェクト

それから、別のブログ「Java Message Queue-ActiveMq in Practice」 、皆さんと一緒に、ActiveMq プロジェクトを 0 から 1 まで開始しました。プロジェクトの開発プロセス中に、ActiveMq について一定の理解がありました:

1。複数の言語とプロトコルでクライアントを作成します。言語: Java、C、C++、C#、Ruby、Perl、Python、PHP。アプリケーションプロトコル: OpenWire、Stomp REST、WS Notice、XMPP、AMQP

2。JMS1.1 および J2EE 1.4 仕様を完全にサポート (永続性、Spring を使用してシステムに簡単に組み込むことができ、Spring 2.0 の機能もサポートします

4)。 JCA 1.5 リソース アダプターを含む一般的な J2EE サーバー (Geronimo、JBoss 4、GlassFish、WebLogic など) のテストに合格しました。構成により、J2EE 1.4

5 と互換性のある商用サーバーに ActiveMQ を自動的にデプロイできます。複数の送信プロトコルをサポートします。 VM 内、TCP、SSL、NIO、UDP、JGroups、JXTA

6. JDBC およびジャーナルによるサポートにより、高速メッセージ永続性が実現

7 高性能クラスター、クライアントサーバー、ポイントを保証するように設計されています。 to-point

8. Ajax のサポート

9. Axis との統合のサポート

10. テストのために組み込み JMS プロバイダーを簡単に呼び出すことができます

次のブログでは、Spring とActiveMq. このブログでは、Spring+JMS+ActiveMQ+Tomcat の実装された Point-to-Point 非同期キュー メッセージと PUB/SUB (パブリッシュ/サブスクライブ) モデルに基づいており、ビジネスは含まれていません。

2. ディレクトリ構造

2.1 プロジェクトディレクトリ

IDE は、jar のダウンロードに関するさまざまなトラブルを回避するために、Maven を使用してプロジェクトを構築し、統合しました。 Spring と 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>

ここのpom.xmlファイルは少し長いので展開しません。 1. Spring コアの依存関係 2. ActiveMq コアとプール (学生がここで jar をインポートすることを選択した場合、以前のブログ パッケージで説明した activemq-all jar を直接インポートできます) 3 . Java サーブレット関連の依存関係

ここで選択する ActiveMq プールの依存関係バージョンは、後続の dtd に関連し、そのバージョンは対応する必要があるため、学生は後で activemq ファイルを構成するときに dtd バージョンの選択に注意する必要があります

2.3 web.xml

web.xml も同様で、Spring 設定ファイル、springMvc の命名、エンコード形式を指定します

<?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 と applicationContext.xml

SpringMVC について特別なことは何もありません。必要な学生は参照できます。見てください:

<?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 は主に Bean をロードするために使用されます。私たちのプロジェクトには特別な Java Bean はないため、パッケージのスキャン パスを示すためにのみ使用されます:

<?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>

ここでは、この設定ファイルについて説明します。上記の設定ファイルを見て理解できる場合は、読み飛ばしてください。学生は ActiveMQ 公式 Web サイトでも閲覧できます。

1. ActiveMq の DTD、関連する設定を宣言する前に、DTD を ActiveMq にインポートする必要があります。そうしないと、Spring はタグの意味を理解できません。

リンクファクトリーのアドレス、ユーザー名、パスワードを設定するために使用される、pom 項目で activemq のバージョンを設定しました。ここで注意する必要があるのは、http 接続の代わりに tcp 接続を選択することです。 jmsTemplate: より重要な設定として、接続ファクトリーがここで指定され、デフォルトのメッセージ送信先、接続期間とメッセージの公開方法が指定されます

3. プロジェクト構造

3.1ProducerService

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);
   }
  });
 }
}

メッセージを送信する必要がある場合は、ProducerService インスタンスの sendMessage メソッドを呼び出して、デフォルトの宛先にメッセージを送信するだけです。

ここでは 2 つの送信方法が提供されています。1 つはデフォルトの宛先に送信する方法、もう 1 つは宛先に基づいてメッセージを送信する方法です。

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

上記は、ActiveMq を統合した Java Message Queue-Spring の詳細な説明です。その他の関連コンテンツについては、PHP 中国語 Web サイト (www.php.cn) に注目してください。