如何：实现 BatchMessageListenerContainer 以批量使用 JMS 队列

Question

我最近在 Spring 集成中遇到了对 JMS 消费者的需求 - 能够消耗大量的突发数据，而不会对我的目标 Oracle 数据库造成太多提交压力。

DefaultMessageListenerContainer 似乎只支持逐条消息事务。

我在谷歌上搜索了解决方案并找到了几个 - 但其中很多都不是通过从 DMLC 继承而是通过克隆和修改相同的原始源代码来实现的 - 这使得它很容易被破坏以防我以后希望移动到更新版本的 spring-jms。此外，被克隆的代码引用了 DMLC 的私有属性，因此必须将其排除在外。为了使这一切正常工作，还需要几个接口和一个自定义消息监听器。总而言之，我感觉不舒服。

那么 - 怎么办？

Answer 1

嗯 - 这是一个简单紧凑的解决方案，完全基于从 DefaultMessageListenerContainer 派生的单个 class。

虽然我只测试了消息驱动通道适配器和 ChainedTransactionManager - 因为这是需要做这样的事情时的基本场景。

这是代码：

package dk.itealisten.myservice.spring.components;

import org.springframework.jms.listener.DefaultMessageListenerContainer;

import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import java.util.ArrayList;
import java.util.Enumeration;

public class BatchMessageListenerContainer extends DefaultMessageListenerContainer {

    public static final int DEFAULT_BATCH_SIZE = 100;

    public int batchSize = DEFAULT_BATCH_SIZE;

    /**
     * Override the method receiveMessage to return an instance of BatchMessage - an inner class being declared further down.
     */
    @Override
    protected Message receiveMessage(MessageConsumer consumer) throws JMSException {
        BatchMessage batch = new BatchMessage();
        while (!batch.releaseAfterMessage(super.receiveMessage(consumer))) ;
        return batch.messages.size() == 0 ? null : batch;
    }

    /**
     * As BatchMessage implements the javax.jms.Message interface it fits perfectly into the DMLC - only caveat is that SimpleMessageConverter dont know how to convert it to a Spring Integration Message - but that can be helped.
     * As BatchMessage will only serve as a container to carry the actual javax.jms.Message's from DMLC to the MessageListener it need not provide meaningful implementations of the methods of the interface as long as they are there.
     */
    protected class BatchMessage implements Message {

        public ArrayList<Message> messages = new ArrayList<Message>();

        /**
         * Add message to the collection of messages and return true if the batch meets the criteria for releasing it to the MessageListener.
         */
        public boolean releaseAfterMessage(Message message) {
            if (message != null) {
                messages.add(message);
            }
            // Are we ready to release?
            return message == null || messages.size() >= batchSize;
        }

        // Below is only dummy-implementations of the abstract methods of javax.jms.Message

        @Override
        public String getJMSMessageID() throws JMSException {
            return null;
        }

        @Override
        public void setJMSMessageID(String s) throws JMSException {

        }

        @Override
        public long getJMSTimestamp() throws JMSException {
            return 0;
        }

        @Override
        public void setJMSTimestamp(long l) throws JMSException {

        }

        @Override
        public byte[] getJMSCorrelationIDAsBytes() throws JMSException {
            return new byte[0];
        }

        @Override
        public void setJMSCorrelationIDAsBytes(byte[] bytes) throws JMSException {

        }

        @Override
        public void setJMSCorrelationID(String s) throws JMSException {

        }

        @Override
        public String getJMSCorrelationID() throws JMSException {
            return null;
        }

        @Override
        public Destination getJMSReplyTo() throws JMSException {
            return null;
        }

        @Override
        public void setJMSReplyTo(Destination destination) throws JMSException {

        }

        @Override
        public Destination getJMSDestination() throws JMSException {
            return null;
        }

        @Override
        public void setJMSDestination(Destination destination) throws JMSException {

        }

        @Override
        public int getJMSDeliveryMode() throws JMSException {
            return 0;
        }

        @Override
        public void setJMSDeliveryMode(int i) throws JMSException {

        }

        @Override
        public boolean getJMSRedelivered() throws JMSException {
            return false;
        }

        @Override
        public void setJMSRedelivered(boolean b) throws JMSException {

        }

        @Override
        public String getJMSType() throws JMSException {
            return null;
        }

        @Override
        public void setJMSType(String s) throws JMSException {

        }

        @Override
        public long getJMSExpiration() throws JMSException {
            return 0;
        }

        @Override
        public void setJMSExpiration(long l) throws JMSException {

        }

        @Override
        public long getJMSDeliveryTime() throws JMSException {
            return 0;
        }

        @Override
        public void setJMSDeliveryTime(long l) throws JMSException {

        }

        @Override
        public int getJMSPriority() throws JMSException {
            return 0;
        }

        @Override
        public void setJMSPriority(int i) throws JMSException {

        }

        @Override
        public void clearProperties() throws JMSException {

        }

        @Override
        public boolean propertyExists(String s) throws JMSException {
            return false;
        }

        @Override
        public boolean getBooleanProperty(String s) throws JMSException {
            return false;
        }

        @Override
        public byte getByteProperty(String s) throws JMSException {
            return 0;
        }

        @Override
        public short getShortProperty(String s) throws JMSException {
            return 0;
        }

        @Override
        public int getIntProperty(String s) throws JMSException {
            return 0;
        }

        @Override
        public long getLongProperty(String s) throws JMSException {
            return 0;
        }

        @Override
        public float getFloatProperty(String s) throws JMSException {
            return 0;
        }

        @Override
        public double getDoubleProperty(String s) throws JMSException {
            return 0;
        }

        @Override
        public String getStringProperty(String s) throws JMSException {
            return null;
        }

        @Override
        public Object getObjectProperty(String s) throws JMSException {
            return null;
        }

        @Override
        public Enumeration getPropertyNames() throws JMSException {
            return null;
        }

        @Override
        public void setBooleanProperty(String s, boolean b) throws JMSException {

        }

        @Override
        public void setByteProperty(String s, byte b) throws JMSException {

        }

        @Override
        public void setShortProperty(String s, short i) throws JMSException {

        }

        @Override
        public void setIntProperty(String s, int i) throws JMSException {

        }

        @Override
        public void setLongProperty(String s, long l) throws JMSException {

        }

        @Override
        public void setFloatProperty(String s, float v) throws JMSException {

        }

        @Override
        public void setDoubleProperty(String s, double v) throws JMSException {

        }

        @Override
        public void setStringProperty(String s, String s1) throws JMSException {

        }

        @Override
        public void setObjectProperty(String s, Object o) throws JMSException {

        }

        @Override
        public void acknowledge() throws JMSException {

        }

        @Override
        public void clearBody() throws JMSException {

        }

        @Override
        public <T> T getBody(Class<T> aClass) throws JMSException {
            return null;
        }

        @Override
        public boolean isBodyAssignableTo(Class aClass) throws JMSException {
            return false;
        }
    }
}

下面的示例展示了如何在 Spring 应用程序上下文中使用它：

<beans xmlns="http://www.springframework.org/schema/beans"
  xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
  xmlns:int="http://www.springframework.org/schema/integration"
  xmlns:jms="http://www.springframework.org/schema/integration/jms"
  xmlns:p="http://www.springframework.org/schema/p"
  xsi:schemaLocation="
    http://www.springframework.org/schema/beans     
    http://www.springframework.org/schema/beans/spring-beans-4.0.xsd
    http://www.springframework.org/schema/integration
    http://www.springframework.org/schema/integration/spring-integration-4.0.xsd
    http://www.springframework.org/schema/integration/jms
    http://www.springframework.org/schema/integration/jms/spring-integration-jms-4.0.xsd">

<!-- Plug in the BatchMessageListenerContainer in a message-driven-channel-adapter -->
<jms:message-driven-channel-adapter container-class="dk.itealisten.myservice.spring.components.BatchMessageListenerContainer"
  acknowledge="transacted"
  channel="from.mq"
  concurrent-consumers="5"
  max-concurrent-consumers="15"
  connection-factory="jmsConnectionFactory"
  transaction-manager="transactionManager"
  destination="my.mq.queue"
  />

<!-- Flow processing the BatchMessages being posted on the "from.mq" channel -->
<int:chain input-channel="from.mq" output-channel="nullChannel">
  <int:splitter expression="payload.messages" />
  <!-- This is where we deal with conversion to spring messages as the payload is now a single standard javax.jms.Message implementation -->
  <int:transformer ref="smc" method="fromMessage"/>
  <!-- And finally we persist -->
  <int:service-activator ref="jdbcPublisher" method="persist"/>
</int:chain>

<!-- Various supporting beans -->

<!-- A bean to handle the database persistance --> 
<bean id="jdbcPersistor" class="dk.itealisten.myservice.spring.components.JdbcPersistor" p:dataSource-ref="dataSource" />

<!-- A bean to handle the conversion that could not take place in the MessageListener as it don't know how to convert a BatchMessage -->
<bean id="smc" class="org.springframework.jms.support.converter.SimpleMessageConverter"/>

<!-- Transaction manager must make sure messages are committed outbound (JDBC) before cleaned up inbound (JMS). -->
<bean id="transactionManager" class="org.springframework.data.transaction.ChainedTransactionManager">
  <constructor-arg name="transactionManagers">
    <list>
      <bean class="org.springframework.jms.connection.JmsTransactionManager" p:connectionFactory-ref="jmsConnectionFactory" />
      <bean class="org.springframework.jdbc.datasource.DataSourceTransactionManager" p:dataSource-ref="dataSource" />
    </list>
  </constructor-arg>
</bean>

如何：实现 BatchMessageListenerContainer 以批量使用 JMS 队列

How to: Implement a BatchMessageListenerContainer for bulk consuming a JMS queue

performance

jms

spring-integration

bulk

consuming