Spring Integration集成框架之JDBC

好久没有更新Blog了，今天突如其来的闲暇发现了一个很好的功能，籍此更新一下。

上一篇基本介绍了Spring Integration，以及其工做模式。也许你会以为它没有特点的东西，本身实现也没有多大难度。是的！本身实现确实是没多大难度，可是本身实现有要具备良好扩展性的仍是不是那么容易的。

咱们的系统常常要和其联系的多个系统一块儿协同工做。他们都操做着数据库的同一张表。如：一个系统向表写数据，另外一个系统定时的扫描新加入的数据，而后把新加入的数据提取出来，作一些处理。而后更新标志或者转移。

这样的场景你遇到过吗？ 反正咱们是有不少的这样的场景。

我以这样的为例作个Demo。

我建立一个这样的数据库。DDL SQL如：

CREATE TABLE PUBLIC.PUBLIC.ATTR_MESSAGE ( 
  ATT_CODE VARCHAR(20) NOT NULL, 
  PARENT_CODE VARCHAR(20), 
  ATT_TEXT VARCHAR(100), 
  SEQ NUMERIC(8, 0), 
  OPT_DATE DATE, 
  MARK VARCHAR(1) DEFAULT 'N', 
  PRIMARY KEY(ATT_CODE) 
);

如上面的表结构，我从别的数据库提取了几个列建立一个表。 MARK就是一个标志列，当新加入的数据为N, 处理后的会置成Y。

Spring Integration JDBC能给你完成几乎全部代码。以下面的Spring配置：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:integration="http://www.springframework.org/schema/integration"
       xmlns:hdbc="http://www.springframework.org/schema/integration/jdbc"
       xmlns:stream="http://www.springframework.org/schema/integration/stream"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
           http://www.springframework.org/schema/beans/spring-beans-3.1.xsd
           http://www.springframework.org/schema/integration
           http://www.springframework.org/schema/integration/spring-integration-2.1.xsd
           http://www.springframework.org/schema/integration/jdbc
		   http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc-2.1.xsd
		   http://www.springframework.org/schema/integration/stream
		   http://www.springframework.org/schema/integration/stream/spring-integration-stream-2.1.xsd">

    <bean id="propertyConfigurer"
          class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer">
        <property name="locations">
            <list>
                <value>classpath:jdbc/jdbc.properties</value>
            </list>
        </property>
    </bean>

    <bean id="dataSource" class="org.springframework.jdbc.datasource.DriverManagerDataSource">
        <property name="driverClassName" value="${jdbc.driverClass}" />
        <property name="url" value="${jdbc.url}" />
        <property name="username" value="${jdbc.username}" />
        <property name="password" value="${jdbc.password}" />
    </bean>

    <bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
        <property name="dataSource" ref="dataSource"/>
    </bean>

    <bean id="jdbcMessageHandler" class="net.dintegration.handler.JdbcMessageHandler" />

    <integration:channel id="jdbcinchannel" />

    <hdbc:inbound-channel-adapter channel="jdbcinchannel" data-source="dataSource"
               query="SELECT ATT_CODE, PARENT_CODE, ATT_TEXT, SEQ, OPT_DATE, MARK FROM ATTR_MESSAGE WHERE MARK = 'N'"
               update="UPDATE ATTR_MESSAGE SET MARK = 'Y' WHERE ATT_CODE IN (:ATT_CODE)">
        <integration:poller fixed-rate="10000">
            <integration:transactional />
        </integration:poller>
    </hdbc:inbound-channel-adapter>

    <integration:service-activator input-channel="jdbcinchannel" ref="jdbcMessageHandler"/>
</beans>

请你注意其中的:

query="SELECT ATT_CODE, PARENT_CODE, ATT_TEXT, SEQ, OPT_DATE, MARK FROM ATTR_MESSAGE WHERE MARK = 'N'" update="UPDATE ATTR_MESSAGE SET MARK = 'Y' WHERE ATT_CODE IN (:ATT_CODE)"

它作做用就是把表ATTR_MESSAGE中MARK=‘N’的数据过滤出来， 放到jdbcMessageHandler中处理，而后按照提取时的 ATT_CODE分别把标志位 MARK置成Y。

如上，咱们只须要编写一个 jdbcMessageHandler处理咱们的数据就好，其余的一切都让Spring Integration为咱们作好了。

public class JdbcMessageHandler  implements MessageHandler {

    private static Log log = LogFactory.getLog(JdbcMessageHandler.class);

    public JdbcMessageHandler() {
    }

    @Override
    public void handleMessage(Message<?> message) throws MessagingException {
        Object obj = message.getPayload();
        //分别按照各类样式输出obj
        if(obj == null) {
            log.info("null");  
        } else if(obj instanceof String) {
            log.info(obj);
        }else if(obj instanceof List) {
            List bean = (List)obj;
            log.info(bean);
        } else {
            log.info(ReflectionToStringBuilder.reflectionToString(message));
        }
    }
}

OK。我向创建的表中插入2条数据, 而后测试。测试类：

public static void main(String[] args) {
        ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("jdbc/jdbcIntegrationContext.xml");
        context.start(); //让线程在这里阻塞，防止JVM退出
    }

测试log如：

11-26 19:27:18 [INFO] [support.DefaultLifecycleProcessor(334)] Starting beans in phase 2147483647 11-26 19:27:19 [INFO] [handler.JdbcMessageHandler(49)] [{ATT_CODE=123456, PARENT_CODE=Root, ATT_TEXT=测试数据, SEQ=1, OPT_DATE=14:17:47, MARK=N}, {ATT_CODE=234567, PARENT_CODE=123456, ATT_TEXT=test, SEQ=2, OPT_DATE=14:20:41, MARK=N}]

很明显它读到了2条数据输出了。请注意，我在Spring中配置的integration:poller fixed-rate="10000"，也就是说每10秒中扫描一次ATTR_MESSAGE表.我再次用一个SQL把刚处理过的数据置成N。如： UPDATE attr_message set mark = 'N'

它也再次输出了日志，如：

11-26 19:30:18 [INFO] [handler.JdbcMessageHandler(49)] [{ATT_CODE=123456, PARENT_CODE=Root, ATT_TEXT=测试数据, SEQ=1, OPT_DATE=14:17:47, MARK=N}, {ATT_CODE=234567, PARENT_CODE=123456, ATT_TEXT=test, SEQ=2, OPT_DATE=14:20:41, MARK=N}]

它又读取了MARK为N的数据。就这样几乎不写任何多余的代码就实现了我上面提到的场景。而咱们须要作的，仅仅写一个MessageHandler处理咱们的数据。

那么他的扩展性呢？

若是你仔细看了，你就发现

<integration:channel id="jdbcinchannel" />

<integration:service-activator input-channel="jdbcinchannel" ref="jdbcMessageHandler"/>

这样的代码在上一篇JMS也曾出现过相似的.如:

<integration:channel id="jmsinchannel"/>

    <integration:channel id="jmsoutchannel"  />
 
    <jms:inbound-channel-adapter id="jmsIn" destination="myTopic" channel="jmsinchannel" jms-template="jmsTemplate">
        <integration:poller fixed-rate="30000"/>
    </jms:inbound-channel-adapter>
 
    <integration:transformer ref="messageTransformer"
                             input-channel="jmsinchannel" output-channel="jmsoutchannel" />
    <integration:service-activator ref="messageHander" input-channel="jmsoutchannel" />

是的，Spring Integration就是经过相似的方式把任何的数据经过管道同样的把数据导向下一个须要的地方。