ActiveMq와의 Spring 통합에 대한 Java 메시지 큐에 대한 자세한 설명

이 글은 주로 Java 메시지 큐-ActiveMq와의 Spring 통합에 대한 자세한 설명을 소개합니다. 편집자는 꽤 좋다고 생각하므로 이제 공유하고 참고용으로 제공하겠습니다. 편집기를 따라 살펴보겠습니다

1. 개요

먼저 Java 메시징 서비스를 함께 살펴보겠습니다. 이전 블로그 "Java Message Queue-JMS 개요"에서 모두를 대상으로 분석했습니다.

1. 메시지 서비스: 두 개 또는 여러 프로그램 간의 결합을 해결하는 데 사용되는 미들웨어, 맨 아래 계층은 자바로 구현.

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 알림, 🎜>3. Spring 지원, ActiveMQ는 Spring을 사용하여 시스템에 쉽게 내장될 수 있으며 Spring 2.0

4의 기능도 지원합니다. JCA 1.5 리소스 어댑터 구성을 통해 J2EE 1.4와 호환되는 모든 상용 서버에 ActiveMQ를 자동으로 배포할 수 있는 J2EE 서버(예: Geronimo, JBoss 4, GlassFish, WebLogic) 테스트

5 . 다중 전송 프로토콜 지원: VM 내, TCP, SSL, NIO, UDP, JXTA

6. JDBC 및 저널을 통해 고속 메시지 지속성 지원

7. 고성능 클러스터링, 클라이언트-서버, 지점 간

8. Ajax 지원

9. >

10. 테스트를 위해 임베디드 JMS 공급자를 쉽게 호출할 수 있습니다

다음 블로그에서는 Spring과 ActiveMq를 통합하는 작업을 진행하겠습니다. Spring+JMS+ActiveMQ+Tomcat은 Point-To-Point 비동기 큐 메시지와 PUB/SUB(publish/subscribe) 모델을 구현하며 간단한 예제이며 어떠한 비즈니스도 포함하지 않습니다.

2. 디렉토리 구조

2.1 프로젝트 디렉토리

IDE에서 선택한 IDEA(누구나 사용 권장) 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를 직접 가져올 수 있습니다. 이 jar 패키지) 3. Java 서블릿 관련 종속성

여기에서 선택한 ActiveMq 풀의 종속성 버전은 후속 dtd와 관련되며 버전이 일치해야 하므로 학생들은 activemq를 구성할 때까지 기다립니다. file.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 공식 웹사이트에서도 볼 수 있습니다.

1. ActiveMq의 DTD, 관련 구성을 선언하기 전에 ActiveMq에서 DTD를 가져와야 합니다. 그렇지 않으면 Spring은 태그의 의미를 이해하지 못합니다.

pom.connectionFactory에서 activemq 버전을 구성했습니다. 링크 팩토리의 주소, 사용자 이름 및 비밀번호를 구성하는 데 사용되는 매우 간단한 구성 항목입니다. 여기서 주목해야 할 것은 tcp를 선택하는 것입니다. http 연결 대신 연결

3. jmsTemplate: 더 중요한 구성, 여기서 연결 팩토리, 기본 메시지 전송 대상, 연결 기간 및 메시지 게시 방법이 지정됩니다

3. 프로젝트 구조

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

메시지 생성자를 서비스로 만듭니다. 메시지를 보내야 하는 경우 ProducerService 인스턴스에서 sendMessage 메서드만 호출하면 됩니다. 기본 대상으로 메시지를 보낼 수 있습니다.

여기에서는 두 가지 전송 방법이 제공됩니다. 하나는 기본 대상으로 보내는 것이고, 다른 하나는 대상을 기준으로 메시지를 보내는 것입니다.

3.2 소비자 서비스

아아앙

因为我们项目中并没有什么业务，所以的话对消息的处理也就是打印输出。我们只需要调用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 중국어 홈페이지(www.php.cn)를 참고해주세요!