ActiveMQ与Spring线程池的整合

转自：http://www.g4studio.org/thread-880-1-1.html

ActiveMQ与Spring线程池的整合在企业级开发中，很多系统间的通信，特别是与外部系统间的通信，往往都是异步的，JMS便是J2EE应用程序中用于处理异步消息传递的接口。为了提高对外部消息的相应，服务器程序中往往利用线程技术来处理接收的消息，线程池的意义在于对这样的一个并行处理机制进行性能上的优化。为了迅速切入正体，这里就不多涉及JMS的内容与池的概念。仅对如何进行ActiveMQ与Spring线程池整合做较为详细的描述。?
引用:

ActiveMQ与Spring线程池整合实例?
CSDN资源下载?仅有47k?

整合步骤?

这里我按照配置流程逐点来描述进行配置的方法。其中MQ整合配置的过程中各bean之间的关系比较多，也比较晕，我用橙红色将他们标记出来，关注标记的几点是十分重要的。?

• 让Spring支持ActiveMQ的配置语法
  

  
  首先我们需要在applicationContext.xml中引入ActiveMQ的配置语法。

  ? ?? ???<amq:destinations>? ?
  ? ?? ?? ?? ?<amq:queue id="jms.log" physicalName="pams.amq" />? ?
  ? ?? ???</amq:destinations>? ?
  ? ?? ???<amq:transportConnectors>? ?
  ? ?? ?? ?? ?<amq:transportConnector uri="tcp://localhost:61616" />? ?
  ? ?? ???</amq:transportConnectors>? ?
  ? ? </amq:broker>? ?
  //配置ConnectionFactory? ?
  ? ? <amq:connectionFactory id="jmsConnectionFactory"??
  ? ?? ???brokerURL="vm://localhost:61616" />? ?
  //配置Queue? ?
  ? ? <amq:queue name="destination" physicalName="pams.amq" />复制代码

  
  
  //配置ActiveMQ Broker<amq:broker useJmx="false" persistent="false">? ?? ?? ?? ?? ? <amq:destinations>? ?? ?? ?? ?? ?? ?? ?? ?<amq:queue id="jms.log" physicalName="pams.amq" />? ?? ?? ?? ?? ? </amq:destinations>? ?? ?? ?? ?? ? <amq:transportConnectors>? ?? ?? ?? ?? ?? ?? ?? ?<amq:transportConnector uri="tcp://localhost:61616" />? ?? ?? ?? ?? ? </amq:transportConnectors>? ?? ???</amq:broker>//配置ConnectionFactory? ?? ???<amq:connectionFactory id="jmsConnectionFactory"? ?? ?? ?? ?? ? brokerURL="vm://localhost:61616" />//配置Queue? ?? ???<amq:queue name="destination" physicalName="pams.amq" />在这里使用了内嵌JVM这种最简单的模式，这样在Spring初始化时ActiveMQ便加载了。?
  对于ActiveMQ的配置，我们还需要配置消息生产者、消息消费者和消息转换者等等部分。不过，在此之前我们需要先将线程池配置进来，之后再进行ActiveMQ余下的配置。?
  
  

  
  
  • 配置Spring线程池
    

    
    
    对于线程池的配置，我们需要配置一个线程池执行器。首先看下配置代码：

    ? ?? ?? ?? ?<property name="corePoolSize" value="2" />? ?
    ? ?? ?? ?? ?<property name="maxPoolSize" value="4"/>? ?
    ? ?? ?? ?? ?<property name="queueCapacity" value="500"/>? ?
    ? ?? ?? ?? ?<property name="keepAliveSeconds" value="300"/>? ?
    ? ?? ?? ?? ?<FONT color=red>
    <property name="rejectedExecutionHandler">? ?
    ? ?? ?? ?? ?? ? <bean ref="threadFactory" /> -->? ?? ???<property name="corePoolSize" value="2" />? ?? ???<property name="maxPoolSize" value="4"/>? ?? ???<property name="queueCapacity" value="500"/>? ?? ???<property name="keepAliveSeconds" value="300"/>? ?? ???<property name="rejectedExecutionHandler">? ?? ?? ?? ?? ? <bean />??复制代码

    
    
    
    
    //消息转换器<bean id="resourceMessageConverter" />

    
    
    • jmsTemplate的配置
      

      
      之后我们需要配置一个由Spring提供的Template，其绑定ConnectionFactory与消息转换器（Converter）。这样在进行消息的生产时我们可以直接使用Spring提供的Template中的方法。

      ? ?? ?? ?? ?? ? </bean>? ?
      ? ?? ?? ?? ?</property>? ?
      ? ?? ?? ?? ?? ?? ?? ? //消息转换器? ?
      ? ?? ?? ?? ?<property name="messageConverter"??
      ? ?? ?? ?? ?? ? ref="resourceMessageConverter" />? ?
      ? ?? ???</bean>??复制代码

      
      
      
      
      <!-- JMS ActiveMQ Config-->? ? <bean id="jmsTemplate"? ?? ?? ?? ?? ? />? ?? ?? ?? ?? ?? ?? ?? ?</bean>? ?? ?? ?? ?? ? </property>? ?? ?? ?? ?? ?? ?//消息转换器? ?? ?? ?? ?? ? <property name="messageConverter"? ?? ?? ?? ?? ?? ?? ?? ?ref="resourceMessageConverter" />? ?? ???</bean>

      
      
      • 消息生产者
        

        
        消息生产者用于生产消息，也就是将消息发送出去。在这里我们使用上面配置好的JmsTemplate来发送发送消息。这个类的代码我就不贴出来了，大家可以在实例中查看。配置消息生产者需要JmsTemplate与消息队列(Queue)。

        ? ?? ?? ?? ?<property name="template">? ?
        ? ?? ?? ?? ?? ? <ref bean="jmsTemplate"></ref>? ?
        ? ?? ?? ?? ?</property>? ?
        ? ?? ?? ?? ?<property name="destination">? ?
        ? ?? ?? ?? ?? ? <ref bean="destination"></ref>? ?
        ? ?? ?? ?? ?</property>? ?
        ? ?? ???</bean>??复制代码

        
        
        
        
        <bean id="resourceMessageProducer"? ?? ?? ?? ?>? ?? ?? ?? ?<property name="template">? ?? ?? ?? ?? ?? ???<ref bean="jmsTemplate"></ref>? ?? ?? ?? ?</property>? ?? ?? ?? ?<property name="destination">? ?? ?? ?? ?? ?? ???<ref bean="destination"></ref>? ?? ?? ?? ?</property>? ? </bean>

        
        
        • 消息接收处理者
          

          
          消息接收者（MDP）使用Spring的MessageListenerAdapter，指定负责处理消息的POJO及其方法名，绑定消息转换器Converter。

          ? ?? ?? ?? ?<property name="messageConverter"? ?ref="resourceMessageConverter" />? ?
          ? ?? ???</bean>? ?复制代码

          
          
          
          
          <bean id="resourceMessageListener"? ?? ?? ?? ?? ? value="addResource" />? ?? ?? ?? ?? ? <property name="messageConverter"? ?ref="resourceMessageConverter" />? ?? ???</bean>

          
          
          • 监听容器的配置
            

            
            监听容器的任务是负责调度MDP，?绑定connectionFactory,Queue和MDP。更重要的一点，这里也是将线程执行器与ActiveMQ整合的切入点。先看下配置文件：

            ? ?? ?? ?? ?<property name="autoStartup" value="true"/>? ??
            ? ?? ?? ?? ?<property name="concurrentConsumers" value="6"/>? ??
            ? ?? ?? ?? ?<property name="destination" ref="destination"/>? ??
            ? ?? ?? ?? ?<property name="messageListener" ref="resourceMessageListener"/>? ??
            ? ?? ?? ?? ?<property name="sessionTransacted" value="true"/>? ??
            ? ?? ?? ?? ?<FONT color=red><property name="taskExecutor" ref="threadPoolExecutor"/> </FONT>? ?
            ? ?? ???</bean>??复制代码

            
            
            
            
            <bean id="resourcelistenerContainer"? ?? ?? ?? ?? ???ref="jmsConnectionFactory" />? ?? ?? ?? ?? ???<property name="autoStartup" value="true"/>? ?? ?? ?? ?? ???<property name="concurrentConsumers" value="6"/>? ?? ?? ?? ?? ???<property name="destination" ref="destination"/>? ?? ?? ?? ?? ???<property name="messageListener" ref="resourceMessageListener"/>? ?? ?? ?? ?? ???<property name="sessionTransacted" value="true"/>? ?? ?? ?? ?? ???<property name="taskExecutor" ref="threadPoolExecutor"/>?? ?? ???</bean>这里通过taskExecutor属性将已经配置好的线程执行器threadPoolExecutor与JMS(ActiveMQ)整合起来。?
            
            

            
            
            • 消息消费者的配置
              

              
              消息消费者是消息的最终处理部分，开始文章里面说说的整合线程的目的，也就是为了使得处理者能够并行的处理接收到的消息。同样这里为了简洁也不将代码贴出来。配置时需要绑定线程池执行器。

              ? ?? ?? ?? ?lazy-init="default" autowire="default" dependency-check="default">? ?
              ? ?? ?? ?? ?<property name="threadPoolExecutor">? ?
              ? ?? ?? ?? ?? ? <ref bean="threadPoolExecutor"></ref>? ?
              ? ?? ?? ?? ?</property>? ?
              ? ?? ???</bean>??复制代码

              
              
              
              
              <bean id="resourceMessageConsumer"? ?? ?? ?? ?? ? abstract="false"? ?? ?? ?? ?? ? lazy-init="default" autowire="default" dependency-check="default">? ?? ?? ?? ?? ? <property name="threadPoolExecutor">? ?? ?? ?? ?? ?? ?? ?? ?<ref bean="threadPoolExecutor"></ref>? ?? ?? ?? ?? ? </property>? ?? ???</bean>至此，我们的整合工作就完成了。在上述文中仅仅简单的描述了配置的步骤，对于原理及各类的属性意义并没有详细的描述。因为网上有不少这样的文章。?
              
              写在最后：?
              我们配置的所有步骤其实都是围绕着Spring展开的，虽然我们完成的是一个ActiveMQ与Spring线程池的整合实例，但是如果换做其他MQ中间件，此法同样是适用的。因为我们整合的层面是在Spring上。