您的位置:首页 > 编程语言 > Java开发

ActiveMQ与Spring的整合使用

2017-08-11 13:39 337 查看
在介绍ActiveMQ特性的时候有一条就是,对spring的支持,ActiveMQ可以很容易内嵌到使用Spring的系统里面去,而且也支持Spring2.0的特性,那么这里就来进行二者的整合与测试使用。

1、生产者

创建maven工程spring-activemq-producer,整一个spring的配置文件放到resource目录下。

在配置文件applicationContext.xml中

1、配置ConnectionFactory;

2、配置spring提供的用于管理ConnectionFactory的ConnectionFactory;

3、配置生成者,Spring为我们提供了JsmTemplate这个工具类,用它可以进行消息发送;

4、配置目的地(有点对点的和一对多的);

<!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供 -->
<bean id="targetConnectionFactory"
class="org.apache.activemq.spring.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://192.168.25.128:61616"/>
</bean>
<!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
<property name="targetConnectionFactory" ref="targetConnectionFactory"/>
</bean>
<!-- 配置生产者 -->
<!-- Spring提供的JMS工具类,它可以进行消息发送、接收等 -->
<bean class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory" ref="connectionFactory"/>
</bean>
<!--这个是队列目的地,点对点的,仅有两个构造函数一个无参一个带String-->
<bean id="queue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="spring.activemq.queue"/>
</bean>
<!--这个是主题目的地,一对多的,仅有两个构造函数一个无参一个带String -->
<bean id="topic" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg value="spring.activemq.topic"/>
</bean>


编写测试类,测试点对点消息发送

public class ActiveMQTest {

/*
* 测试点对点:生产者
*/
@Test
public void test(){
ApplicationContext ac = new ClassPathXmlApplicationContext("classpath:applicationContext-activemq.xml");
//获取生产者
JmsTemplate jmsTemplate = ac.getBean(JmsTemplate.class);
//获取消息发送目的地,由于配置了两个目的地所以需要根据id来取,这里测试的是点对点的,测试发布/订阅要使用id为"topic"
Destination queue = (Destination) ac.getBean("queue");
//发送消息,需要传入生产者
jmsTemplate.send(queue,new MessageCreator() {

@Override
public Message createMessage(Session session) throws JMSException {
TextMessage message = session.createTextMessage("发送了一个消息");
return message;
}
});
}
}


运行方法后页面查看如下





发现一个挂起的消息(该消息是被持久化的,被消费时消失),一个入队消息。

2、消费者

创建另外一个maven工程spring-activemq-consumer,整一个spring的配置文件放到resource目录下。

在配置文件applicationContext.xml中

1、配置ConnectionFactory;

2、配置spring提供的用于管理ConnectionFactory的ConnectionFactory;

3、配置目的地(有点对点的和一对多的),到这都是跟配置生产者一致;

4、编写消息监听器,MyMessageListener(名字随意)实现MessageListener接口。

public class MyMessageListener implements MessageListener{

@Override
public void onMessage(Message message) {
TextMessage textMessage = (TextMessage)message;
try {
//获取消息信息内容
String text = textMessage.getText();
//打印信息
System.out.println(text);
} catch (JMSException e) {
e.printStackTrace();
}
}
}


5、配置文件中配置自定义消息监听器;

6、配置消息监听容器。

<!-- 真正可以产生Connection的ConnectionFactory,由对应的 JMS服务厂商提供 -->
<bean id="targetConnectionFactory"
class="org.apache.activemq.spring.ActiveMQConnectionFactory">
<property name="brokerURL" value="tcp://192.168.25.128:61616"/>
</bean>
<!-- Spring用于管理真正的ConnectionFactory的ConnectionFactory -->
<bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
<!-- 目标ConnectionFactory对应真实的可以产生JMS Connection的ConnectionFactory -->
<property name="targetConnectionFactory" ref="targetConnectionFactory"/>
</bean>

<!--这个是队列目的地,点对点的,仅有两个构造函数一个无参一个带String-->
<bean id="queue" class="org.apache.activemq.command.ActiveMQQueue">
<constructor-arg value="Spring.activemq.queue"/>
</bean>
<!--这个是主题目的地,一对多的,仅有两个构造函数一个无参一个带String -->
<bean id="topic" class="org.apache.activemq.command.ActiveMQTopic">
<constructor-arg value="Spring.activemq.topic"/>
</bean>
<!-- 配置监听器 -->
<bean id="messageListener" class="com.itheima.activemq.MyMessageListener"></bean>
<!-- 消息监听容器 -->
<bean class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="connectionFactory"/>
<!--这里测试是点对点的,使用发布/订阅的话使用topic-->
<property name="destination" ref="queue"/>
<property name="messageListener" ref="messageListener"/>
</bean>


注意:消费者接收消息目的地必须跟生产者发送消息目的地名称一致,不然不在一个频道是接收不到消息的。

测试接收消息,由于前面发送了一个消息,这里是应该能接收到的。

public class ConsumerTest {
@Test
public void testConsumer() throws IOException{
ApplicationContext ac =
new ClassPathXmlApplicationContext("classpath:applicationContext-activemq.xml");
System.in.read();
}
}


注:System.in.read()表示按下回车程序才会继续往下执行,否则是处于一直能接收消息的状态,生成者发送几条消息这就接收几条(也就会打印几条生产者发送的消息的内容 “发送了一个消息“)。

控制台输出:
log4j:WARN No appenders could be found for logger (org.springframework.core.env.StandardEnvironment).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
发送了一个消息


查看页面:



能看到被挂起的消息被消费了,所以有一条出队消息,一个消费者。这里测试的是点对点的,关于发布/订阅形式的上面改一下配置文件,以及测试方法中发布消息目的地改为”topic”即可,注释里面已经写过就不再写测试了。
内容来自用户分享和网络整理,不保证内容的准确性,如有侵权内容,可联系管理员处理 点击这里给我发消息
标签: