Spring配置activemq异步消息监听器

<!-- 创建工厂连接 -->
    <bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL"
            value="failover:(tcp://127.0.0.1:61616)?initialReconnectDelay=100" />
        <property name="useAsyncSend" value="false" />
        <property name="dispatchAsync" value="true" />
    </bean>
    <bean id="pooledJmsConnectionFactory"
        class="org.springframework.jms.connection.CachingConnectionFactory">
        <constructor-arg ref="targetConnectionFactory" />
    </bean>
    
    <bean id="queueDestinationIng" class="org.apache.activemq.command.ActiveMQQueue">
        <constructor-arg>
            <value>order.ing</value>
        </constructor-arg>
    </bean>
    
    <!-- 消息监听器 -->
    <bean id="ingMessageListener" class="com.bypay.listener.IngListener" />

    <!-- 消息监听容器 -->
    <bean id="jmsContainerIng"
        class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="pooledJmsConnectionFactory" />
        <property name="destination" ref="queueDestinationIng" />
        <property name="messageListener" ref="ingMessageListener" />
        <property name="sessionTransacted" value="false"></property>
        <property name="concurrency" value="10-15"></property>
    </bean>

在上述配置中DefaultMessageListenerContainer主要的属性列表如下：

    messageListener: 消息侦听器，必选属性。

 

    taskExecutor: 任务调度器，可以使用线程池开并发的消费消息。如果开发者不指定，spring将会采用默认的TaskExecutor(SimpleAsyncTaskExecutor,类似于CachedThreadPool)。

 

    concurrentConsumers: 消费者的最大个数，因为在spring中messageListener实例是单例的，比如上文中的orderListener(备注：它实现了MessageListener接口)，所以spring-jms不能自作主张的创建多个messageListener实例来并发消费。所以spring在内部，创建了多个MessageConsumer实例，并使用consumer.receive()方法以阻塞的方式来获取消息，当获取消息后，在执行messageListener.onMessage()方法；concurrentConsumers属性就是为了指定spring内部可以创建MessageConsumer的最大个数；当messageConsumer实例被创建后，将会封装在一个Runner接口并交给taskExecutor来调度；如果consumer在一直没有收到消息，则会被置为“idle”并从consumer列表中移除；如果所有的consumer都处于active状态，则会创建新的consumer实例直到达到maxConcurrentConsumers个数上限。通常taskExecutor的线程池容量稍大于concurrentConsumer。

 

    maxMessagesPerTask: 每个consumer所消费的消息个数，因为每个consumer都会独占一个Thread[consumer.receive()是阻塞的]，当consumer消费maxMessagesPerTask个消息后，它就会退出线程，由taskExecutor重新调度。

 

    receiveTimeout: 内部的consumer在receive方法中阻塞的时间。默认为1秒。

 

    recoveryInterval:  当消息消费时，底层connection异常而无法继续，listener需要等待恢复的时间间隔。默认为5000ms。

 

    concurrency: “concurrentConsumers”与“maxConcurrentConsumers”两个参数的简写方式，格式为“5-10”，则表示concurrentConsumers为5，maxConcurrentConsumers为10。

 

    sessionTransacted: Session是否为事务类型。默认为false。

 

    messageSelector: 消息选择器。如果你希望此listener只接受某种特性的消息，可以通过指定selector的方式来过滤消息。

 

    pubSubDomain: 此消费通道是否为Topic，默认为“false”。所有与Topic有关的属性，只有在pubSubDomain为true的情况下生效。

 

    pubSubNoLocal: 对于Topic而言，此消费者是否消费本地消息。所谓本地消息，就是当Consumer与Producer公用底层一个Connection时，那么Producer发送的消息，相对于此Consumer就是本地消息。在pubSubDomain为true时有效。

 

    subscriptionDurable: 是否为“耐久性”订阅者。在pubSubDomain为true时有效。默认为false。

 

    durableSubscriptionName: 耐久订阅者名称，每个clientId下可以有多个耐久订阅者，但是他们必须有不同的名字。默认为className。

 

    errorHandler: 当listener.onMessage方法抛出异常时，异常该如何处理。

 

    autoStartup: 消费者是否自动启动，默认为true，那么在messageContainer实例化后，将会启动consumer(即调用Connection.start())；如果为false，那些开发者需要在合适的时机手动启动。

    clientId: 对于Topic订阅者而言，此参数必备。

 

    sessionAcknowledgeMode: ACK MODE,默认为AUTO。spring-jms做了一件非常遗憾的事情，如果指定了sessionTransacted为true，那么在调用listener.onMessage()方法之后，则会立即提交事务(session.commit())，即使开发者使用了sessionAwareMessageListener，所以开发者无法实现基于事务的“批量”确认机制。如果开发者指定为CLIENT_ACK，那么spring-JMS将会在onMessage方法返回后立即调用message.acknowlege()方法，所以开发者自己是否确认以及何时确认，将没有意义，如果不希望spring来确认消息，只能在onMessage方法中通过抛出异常的方式。 其中“1”表示AUTO_ACKNOWLEDGE,“2”为CLIENT_ACKNOWLEDGE = 2，“3”为 DUPS_OK_ACKNOWLEDGE = 3。