activemq spring 集成与测试

时间:2021-12-12 13:05:38

1.下载安装activemq

2.pom依赖配置

3.spring配置

4.生产消息,消费消息(同步消费),监听消息(异步消费)

4.测试

5.参考博客

http://www.cnblogs.com/leiOOlei/p/5075402.html

1.activemq 官网地址

http://activemq.apache.org/getting-started.html

2.大小

60m 编译文件

300m 源文件

3.操作系统

Windows: Windows XP SP2, Windows 2000, Windows Vista, Windows 7.

Unix: Ubuntu Linux, Powerdog Linux, MacOS, AIX, HP-UX, Solaris.

4.环境

jre 编译环境 1.7

maven  3.0.0

5.activemq 安装

a.下载包

http://activemq.apache.org/download-archives.html

b.解压即可

tar -zxvf apache-activemq-5.9.1-bin.tar.gz

6.启动 activemq

./activemq console

或者

bin\activemq start

7.打开管理页

http://127.0.0.1:8161/admin/

Login: admin

Passwort: admin

8.查看启动日志

tail -f ./data/activemq.log

9.停止 activemq

./activemq stop

10.添加队列,并发布消息

11.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://maven.apache.org/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>activemq</groupId>
<artifactId>activemq</artifactId>
<packaging>war</packaging>
<version>0.0.1-SNAPSHOT</version>
<name>activemq Maven Webapp</name>
<url>http://maven.apache.org</url> <properties>
<springframework>5.0.4.RELEASE</springframework>
<slf4j.version>1.7.7</slf4j.version>
<logback.version>1.1.2</logback.version>
</properties> <dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>3.8.1</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>${springframework}</version>
<scope>provided</scope>
</dependency>
<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>
<!-- 导入java ee jar 包 -->
<dependency>
<groupId>javax</groupId>
<artifactId>javaee-api</artifactId>
<version>7.0</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> <!-- logback -->
<dependency>
<!--主要介绍的是这个jar包,这个包是负责logback随着项目启动的jar包-->
<groupId>org.logback-extensions</groupId>
<artifactId>logback-ext-spring</artifactId>
<version>0.1.4</version>
</dependency>
<!-- slf4j start -->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>${slf4j.version}</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>jcl-over-slf4j</artifactId>
<version>${slf4j.version}</version>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-access</artifactId>
<version>${logback.version}</version>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-core</artifactId>
<version>${logback.version}</version>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>${logback.version}</version>
</dependency>
<!-- slf4j end --> </dependencies>
<build>
<finalName>activemq</finalName>
</build>
</project>

12.spring 配置 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.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-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/mvc
http://www.springframework.org/schema/mvc/spring-mvc-4.1.xsd
http://www.springframework.org/schema/jms
http://www.springframework.org/schema/jms/spring-jms-4.1.xsd
http://activemq.apache.org/schema/core
http://activemq.apache.org/schema/core/activemq-core-5.12.1.xsd"
> <amq:connectionFactory id="amqConnectionFactory"
brokerURL="tcp://192.168.64.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>bonade.q1</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.guo.activemq.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>

13.生产者,消费者

package com.guo.activemq;

import javax.annotation.Resource;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session; import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.stereotype.Service; @Service
public class ProducerService { @Resource(name="jmsTemplate")
private JmsTemplate jmsTemplate; /**
* 向指定队列发送消息
*/
public void sendMessage(Destination destination, final String msg) {
System.out.println("向队列" + 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.getDefaultDestination().toString();
System.out.println("向队列" +destination+ "发送了消息------------" + msg);
jmsTemplate.send(new MessageCreator() {
public Message createMessage(Session session) throws JMSException {
return session.createTextMessage(msg);
}
}); } }
package com.guo.activemq;

import javax.annotation.Resource;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.TextMessage; import org.springframework.jms.core.JmsTemplate;
import org.springframework.stereotype.Service; @Service
public class ConsumerService { @Resource(name="jmsTemplate")
private JmsTemplate jmsTemplate; /**
* 接收消息
*/
public TextMessage receive(Destination destination) {
TextMessage tm = (TextMessage) jmsTemplate.receive(destination);
try {
System.out.println("从队列" + destination.toString() + "收到了消息:\t"
+ tm.getText());
} catch (JMSException e) {
e.printStackTrace();
} return tm; } }

14.添加监听

变下监听类

package com.guo.activemq;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage; public class QueueMessageListener implements MessageListener { @Override
public void onMessage(Message message) { TextMessage tm = (TextMessage) message;
try {
System.out.println("QueueMessageListener监听到了文本消息:\t"
+ tm.getText());
//do something ...
} catch (JMSException e) {
e.printStackTrace();
}
} }

在 activemq.xml中添加如下

 <!-- 配置消息队列监听者(Queue) -->
<bean id="queueMessageListener" class="com.guo.activemq.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>

14.测试

package activemq;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import com.guo.activemq.ProducerService; @RunWith(SpringJUnit4ClassRunner.class) //使用junit4进行测试
@ContextConfiguration(locations={"classpath:spring.xml"}) //加载配置文件
public class ProducerTest { @Autowired
private ProducerService producerService; @Test
public void send() {
producerService.sendMessage("我来了。");
} }
package activemq;

import org.apache.activemq.command.ActiveMQQueue;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import com.guo.activemq.ConsumerService; @RunWith(SpringJUnit4ClassRunner.class) //使用junit4进行测试
@ContextConfiguration(locations={"classpath:spring.xml"}) //加载配置文件
public class ConsumerTest { @Autowired
private ConsumerService consumerService;
@Autowired
private ActiveMQQueue demoQueueDestination; @Test
public void receive() {
consumerService.receive(demoQueueDestination);
} }