Sping+ActiveMQ整合

摘要:
通过前面的ActiveMQ简介,我对ActiveMQ有了大致的了解。本文描述了如何通过Spring继承ActiveMQ,以更有效和灵活地使用ActiveMQ。Spring和ActiveMQ集成需要在项目中包含以下jar包:activeio-core-3.1.4.jar、ActiveMQ-all-5.13.2.jar、ActiveMQ-pool-5.13.2.jar。SingleConnectionFactory将始终为建立JMS服务器链接的请求返回相同的链接,并将忽略Connection的close方法调用。Spring提供的ConnectionFactory仅由Spring用于管理ConnectionFactory。真正生成到JMS服务器的链接的ConnectionFactory必须由JMS服务提供程序提供,并且必须将其注入到Spring提供的ConnectionFactory中。生产者负责生成消息并将其发送到JMS服务器。

通过前一篇《ActiveMQ简述》大概对ActiveMQ有了一个大概的认识。本篇所阐述的是怎样通过Spring继承ActiveMQ进而更有效、更灵活的运用ActiveMQ.

Spring和ActiveMQ整合须要在项目中包括下面这几个jar包(缺一不可):activeio-core-3.1.4.jar,activemq-all-5.13.2.jar,activemq-pool-5.13.2.jar。commons-pool2-2.4.2.jar,这些jar能够在ActiveMQ的安装包中的/lib/optional/下找到,或者从这里下载。

配置ConnectionFactory

ConnectionFactory是用于产生到JMSserver的链接的,Spring为我们提供了多个ConnectionFactory,有SingleConnectionFactory和CachingConnectionFactory。

SingleConnectionFactory对于建立JMSserver链接的请求会一直返回同一个链接,而且会忽略Connection的close方法调用。

CachingConnectionFactory继承了SingleConnectionFactory,所以它拥有SingleConnectionFactory的全部功能,同一时候它还新增了缓存功能。能够缓存Session, MessageProducer和MessageConsumer。

Spring提供的ConnectionFactory仅仅是Spring用于管理ConnectionFactory的,真正产生到JMSserver链接的ConnectionFactory还得是JMS服务厂商提供的,而且须要把它注入到Spring提供的ConnectionFactory中。这里使用ActiveMQ提供的ConnectionFactory,所以定义例如以下(10.10.195.187是博主的測试ip地址):

    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://10.10.195.187:61616" />
    </bean>
    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
        <property name="targetConnectionFactory" ref="targetConnectionFactory"/>
    </bean>

ActiveMQ为我们提供了一个PooledConnectionFactory。通过往里面注入一个ActiveMQConnectionFactory能够用来将Connection, Session和MessageProducer池化,这样能够大大的降低我们的资源消耗。当使用PooledConnectionFactory时。我们在定义一个ConnectionFactory时这样定义:

    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://10.10.195.187:61616" />
    </bean>
    <bean id="poolConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory"  >
        <property name="connectionFactory" ref="targetConnectionFactory" />
        <property name="maxConnections" value="10"/>
    </bean>
    <bean id="connectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">
        <property name="targetConnectionFactory" ref="poolConnectionFactory"/>
    </bean>

这里也能够去掉spring提供的SingleConnectionFactory,相似这样:

    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://10.10.195.187:61616" />
    </bean>
    <bean id="connectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory"  >
        <property name="connectionFactory" ref="targetConnectionFactory" />
        <property name="maxConnections" value="10"/>
    </bean>

配置消息发送(生产)者

配置好ConnectionFactory之后我们就须要配置生产者。生产者负责生产消息并发送到JMSserver。这通常相应的是我们的一个业务逻辑服务实现类。可是我们的服务实现类是怎么进行消息的发送的呢?这一般是利用Spring为我们提供的JmsTemplate类来实现,所以配置生产者事实上最核心的就是配置进行消息发送的JmsTemplate。

对于消息发送者而言,它在发送消息的时候要知道自己该往哪里发。为此,我们在定义JmsTemplate的时候须要往里面注入一个Spring提供的ConnectionFactory对象。

spring配置文件里加入:

    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="poolConnectionFactory"/>
    </bean>

在Java相关处理文件里加入(这里用的是@Inject注解,当然也能够用@Autowired):

    @Inject
    @Named("jmsTemplate")
    private JmsTemplate jmsTemplate;

这样就能够通过jmsTemplate对象来处理相关信息。譬如:

    jmsTemplate.convertAndSend("sqlQueue",sql);

真正利用JmsTemplate进行消息发送的时候,我们须要知道消息发送的目的地,即Destination。在Jms中有一个用来表示目的地的Destination接口,它里面没有不论什么方法定义,仅仅是用来做一个标志而已。

当我们在使用JmsTemplate进行消息发送时没有指定destination的时候将使用默认的Destination。

默认Destination能够通过在定义jmsTemplate bean对象时通过属性defaultDestination或defaultDestinationName来进行诸如。defaultDestinationName对于的就是一个普通字符串。在ActiveMQ中实现了两种类型的Destination。一个是点对点的ActiveMQQueue,还有一个就是支持订阅-公布模式的ActiveMQTopic。

    <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg>
            <value>sqlQueue</value>
        </constructor-arg>
    </bean>
    <bean id="topicDestination" class="org.apache.activemq.command.ActiveMQTopic">
        <constructor-arg>
            <value>topic</value>
        </constructor-arg>
    </bean>

对面上面的那个样例能够在Java程序中加入:

    @Inject
    @Named("queueDestination")
    private Destination queueDestination;

进而【jmsTemplate.convertAndSend(“sqlQueue”,sql);】能够改为【jmsTemplate.convertAndSend(queueDestination,sql);】
也能够这样改动jmsTemplate:

    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="defaultDestination" ref="queueDestination"/>
    </bean>

或者:

    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="defaultDestinationName" value="sqlQueue"/>
    </bean>

进而在java程序中这样调用:jmsTemplate.convertAndSend(sql);


配置消息接收(消费)者

生产者往指定目的地Destination发送消息后,接下来就是消费者对指定目的地的消息进行消费了。

那么消费者是怎样知道有生产者发送消息到指定目的地Destination了呢?这是通过Spring为我们封装的消息监听容器MessageListenerContainer实现的,它负责接收信息,并把接收到的信息分发给真正的MessageListener进行处理。

每一个消费者相应每一个目的地都须要有相应的MessageListenerContainer。

对于消息监听容器而言,除了要知道监听哪个目的地之外。还须要知道到哪里去监听。也就是说它还须要知道去监听那个JMSserver,这是通过在配置MessageConnectionFactory的时候往里面注入一个ConnectionFactory来实现的。

所以我们在配置一个MessageListenerContainer的时候有三个属性必须指定,一个是表示从哪里监听的ConnectionFactory,一个是表示监听什么的Destination。一个是接收到消息以后进行消息处理的MessageListener.

Spring为我们听过了两种类型的MessageListenerContainer:SimpleMessageListenerContainer和DefaultMessageListenerContainer。MessageListenerContainer:SimpleMessageListenerContainer会在一開始的时候就创建一个会话Session和消费者Consumer,而且会适用标准的JMS的MessageConsumer.setMessageListener()方法注冊监听器让JMS提供调用监听器的回调函数。

它不会动态的适应执行时须要和參与外部的事务管理。兼容性方面。它很接近于独立的JMS规范。但一般不兼容J2EE的JMS限制。大多数情况下,我们还是使用DefaultMessageListenerContainer。跟MessageListenerContainer:SimpleMessageListenerContainer想比。它会动态的适应执行时的需求。而且能够參与外部的事务管理。

它很好的平衡了JMS提供者要求低,先进功能如事务參与和兼容J2EE环境。

spring配置文件里加入:

    <bean id="queueDestination" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg>
            <value>sqlQueue</value>
        </constructor-arg>
    </bean>
    <bean id="listenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="poolConnectionFactory" />
        <property name="messageListener" ref="jmsQueueReceiver" />
        <property name="destination" ref="queueDestination" />
    </bean>

当中的jmsQueueReceiver代码例如以下:

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

import org.springframework.stereotype.Component;

@Component("jmsQueueReceiver")
public class JmsQueueReceiver implements MessageListener
{
    @Override
    public void onMessage(Message messgae)
    {
        if(messgae instanceof TextMessage)
        {
            TextMessage textMessage = (TextMessage) messgae;
            try
            {
                String text = textMessage.getText();
                System.out.println("######################["+text+"]######################");
            }
            catch (JMSException e)
            {
                e.printStackTrace();
            }
        }
    }
}

事务管理

Spring提供了一个JmsTransactionManager用于对JMS ConnectionFactory做事务管理。

这将同意JMS应用利用Spring的事务管理特性。JmsTransactionManager在执行本地资源事务管理时将从指定的ConnectionFactory绑定一个ConnectionFactory/Session这种配对到线程中。JmsTemplate会自己主动检測这种事务资源,并对他们进行相应的操作。

在J2EE环境中,ConnectionFactory会池化Connection和Session,这样这些资源将会在整个事务中被有效地反复利用。在一个独立的环境中,使用Spring的SingleConnectionFactory时全部的事务将公用一个Connection,可是每一个事务将保留自己独立的Session.

JmsTemplate能够利用JtaTransactionManager和能够进行分布式的JMS ConnectionFactory处理分布式事务。

在Spring整合JMS的应用中,假设我们要进行本地的事务管理的话很easy,仅仅须要在定义对于的消息监听容器时指定其sessionTransacted属性为true(默觉得false):

    <bean id="listenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="messageListener" ref="jmsQueueReceiver" />
        <property name="destination" ref="queueDestination" />
        <property name="sessionTransacted" value="true"/>
    </bean>

这样JMS在进行消息监听的时候就会进行事务控制,当在接收消息时监听器执行失败时JMS就会对接收到的消息进行回滚。

这里须要注意的是对于其它操作如数据库等訪问不属于该事务控制。

能够在JmsQueueReceiver中改动一下代码从而检測到发送异常时是否会进行事务回滚:

@Component("jmsQueueReceiver")
public class JmsQueueReceiver implements MessageListener
{
    @Override
    public void onMessage(Message messgae)
    {
        if(messgae instanceof TextMessage)
        {
            TextMessage textMessage = (TextMessage) messgae;
            try
            {
                String text = textMessage.getText();
                System.out.println("######################["+text+"]######################");
                if(true)
                {
                    throw new RuntimeException("Error");
                }
            }
            catch (JMSException e)
            {
                e.printStackTrace();
            }
        }
    }
}

你能够通过向Destination发送一条信息,通过JmsQueueReceiver处理后。发送异常进而事务混滚。能够通过http://10.10.195.187:8161/admin/queues.jsp查看相关信息(ip地址是博主的ActiveMQserver所在id,依据实际情况改动)。

假设想要接收消息和数据库訪问处于同一事务中,那么我们就能够配置一个外部的事务管理同一时候配置一个支持外部事务管理的消息监听容器(如DefaultMessageListenerContainer)。要配置这样一个參与分布式事务管理的消息监听容器,我们能够配置一个JtaTransactionManager。当然底层的JMS ConnectionFactory须要能够支持分布式事务管理,并正确地注冊我们的JtaTransactionManager。这样消息监听器进行消息接收和相应的数据库訪问就会处于同一数据库控制下,当消息接收失败或数据库訪问失败都会进行事务回滚操作。

    <bean id="listenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="messageListener" ref="jmsQueueReceiver" />
        <property name="destination" ref="queueDestination" />
        <property name="sessionTransacted" value="true"/>
        <property name="transactionManager" ref="jtaTransactionManager"/>
    </bean>
    <bean id="jtaTransactionManager" class="org.springframework.transaction.jta.JtaTransactionManager"/>

当给小时监听容器指定了transactionManager时,消息监听容器将忽略sessionTransacted的值。

到这里大概的Spring+ActiveMQ整合告一段落。全部代码经博主亲測,确保有效性及正确性。

如有疑问。可在下方留言。


參考资料:
1. Spring整合JMS(一)——基于ActiveMQ实现
2. Spring+ActiveMQ+Mysql 配置JMS
3. ActiveMQ简述

免责声明:文章转载自《Sping+ActiveMQ整合》仅用于学习参考。如对内容有疑问,请及时联系本站处理。

上篇程序的思考之中的一个个能够检验你指针理解功力的程序【Android】自己定义相机的实现(支持连续拍照、前后摄像头切换、连续对焦)下篇

宿迁高防,2C2G15M,22元/月;香港BGP,2C5G5M,25元/月 雨云优惠码:MjYwNzM=

相关文章

Spring3:AOP

转自原文 Spring3:AOP 一、基础概念 AOP AOP(Aspect Oriented Programming),即面向切面编程,可以说是OOP(Object Oriented Programming,面向对象编程)的补充和完善。 OOP引入封装、继承、多态等概念来建立一种对象层次结构,用于模拟公共行为的一个集合。不过 OOP允许开发者定义纵...

@Autowired 与@Resource的区别(转)

原文地址:https://blog.csdn.net/weixin_40423597/article/details/80643990 1、@Autowired与@Resource都可以用来装配bean. 都可以写在字段上,或写在setter方法上。 2、@Autowired默认按类型装配(这个注解是属业spring的),默认情况下必须要求依赖对象必须存在,...

Spring IOC的配置使用

1.1.1 XML配置的结构一般配置文件结构如下: <beans> <import resource=”resource1.xml” /> <bean id=”bean1” class=””></bean> <bean name=”bean2” class=””>...

ActiveMQ消息队列和SignalR之日志实时监控及警报小实例

主要技术: log4net-生成日志。 ActiveMQ-生成日志的时候发送消息,并实时监控日志。 SignalR-将ActiveMQ监控的日志实时显示到浏览器上,而不用刷新浏览器。 小实例介绍: 左侧命名为系统一,右侧命名为系统二 系统一是生成日志的小工具,系统二根据生成的日志实时显示数据,如果ERROR级别的日志超过50条则实时显示警报。 系统一主要代...

带日期的bean转为json(bean-&gt;JSON)

示例代码: JsonBean bean = new JsonBean();bean.setName("NewBaby");bean.setAge(1);bean.setBorn(new Date());jo = JSONObject.fromObject(bean);System.out.println("bean->json:" + jo.toStr...

SpringBoot启动原理解析

1. 启动任何一个Spring Boot项目,都会用到如下类 @SpringBootApplication public class Application { public static void main(String[] args) { SpringApplication.run(Application.class, args...