mule in action翻译19 : 3.7  使用数据库

 

 

     有时数据库是不同应用之间的隐含集成方式。每个主流的开发平台都对数据库交互提供丰富的支持。因此 ，数据库比应用活的更久并不是什么稀罕事。如果你正在使用一个java应用程序，很可能使用了数据库抽象层（实现可能是 Hibernate或 spring jdbc）。这种情况下，在组件中使用这些类库是行得通的。如果你在使用一个遗留数据库或与一个不提供本地java访问方式的应用集成，那么JDBC传输就是访问数据库的有效的方式。

 

    数据库方面最近的发展趋势是NoSQL,它提供另一个非关系型数据解决方案。这些技术往往会使用云计算和分布式计算基础设施以方便横向扩展和处理大数据。

 

    本节，将以学习JDBC传输开始，首先看如何使用JDBC inbound endpoint执行查询。然后学习使用 JDBC outbound endpoint执行插入。我们也将学习下 MongoDB: 一个面向文档的，NoSql数据库。

 

3.7.1 使用 JDBC inbound endpoint 执行查询

    来看如何配置 JDBC传输。表3.7 展示了一些常见配置属性。 dataSource-ref 特别重要，这是你对你配置的数据源的引用，你将使用它访问数据库。

 

  你已经使用过JDBC inbound endpoint 执行数据库查询。它将生成数据，传输给组件和outbound endpoint。来看他们是怎么工作的。 Prancing Donkey 使用 MySQL作为他们网上商城的数据库。网络订单通过一个第三方服务来执行。 Prancing Donkey 周期性的轮询这个服务的API，并更新他们的MySQL数据库中订单的状态。 有时候，订单没有得到及时执行。出现这种情况时，Prancing Donkey员工需要联系订单执行服务， 查出卡住的原因。为了出现这种情况时能给Prancing Donkey提示以进行操作，他们配置了一个mule 流， 来轮询数据库中超过一天没有被执行的订单，然后向一个JMS topic发送一个警告。像下面列表展示的一样。 图示见3.18 

 Listing 3.22 Querying a table every hour and sending the results to a JMS topic 

 

<!--注释1 配置 JDBC 数据源-->
 <spring:bean id="dataSource"
           class="org.springframework.jdbc.datasource.DriverManagerDataSource">
    <spring:property name="driverClassName"  value="${jdbc.driver}"/>
    <spring:property name="url" value="${jdbc.url}"/>
    <spring:property name="username" value="${jdbc.username}"/>
    <spring:property name="password" value="${jdbc.password}"/>
</spring:bean>
<!--注释2 配置jdbc连接器-->
<jdbc:connector name="jdbcConnector" dataSource-ref="dataSource">
<!--注释3 定义查询sql-->
    <jdbc:query key="stuckOrderQuery"
          value="SELECT id FROM orders WHERE TIMESTAMP(timestamp)
           < {fn TIMESTAMPADD( SQL_TSI_DAY, -1, CURRENT_TIMESTAMP)}"/>
</jdbc:connector>
<flow name="pollForStuckOrders">
    <!--每一小时执行一次查询-->
    <jdbc:inbound-endpoint pollingFrequency="3600000"
                                   queryKey="stuckOrderQuery"
                              connector-ref="jdbcConnector"/>
    <jms:outbound-endpoint topic="orders.status.stuck"/>
</flow>

 

注释1处 配置了一个Spring 数据源，其中的一些变量会从一个properties文件读入。

注释2处 引用了这个数据源。

注释3处 定义的查询会使用这个的数据源。

 

如果你仔细观察，你会发现你转义了查询中小于号（？？？），以避免mule启动时XML解析出问题。

这个查询会为每个在上一小时中插入数据库的并在执行中卡住的订单返回一条记录。结果集的每个记录，将作为一条单独的消息发送到JMS话题。消息的payload是  java.util.Map的一个实例。这个Map 使用转换器可以转换为别的对象类型，比如 POJO 或 XML.  map 的key是结果集中的列名，key对应的值是结果集中每行的值。

 

让我们来看看如何使用JDBC outbound endpoint将行插入到数据库。

 

3.7.2 使用 JDBC outbound endpoint 执行插入

 

   JDBC传输允许你使用 outbound endpoint把数据插入到一个表中。回想列表3.12 ，使用FTP

outbound endpoint发送产品数据到零售商。现在修改这个流，增加把商品数据添加到数据库。

 Listing 3.23 Using a JDBC outbound endpoint to insert rows into a table 

 

<!--注释1 定义JDBC连接器 和 sql语句-->
 <jdbc:connector name="jdbcConnector" dataSource-ref="dataSource">
    <jdbc:query key="productInsert"
              value="INSERT INTO products VALUES
                (0,#[payload.name],#[payload.acv],
                #[payload.cost],#[payload.description])"/>
</jdbc:connector>
<flow name="ftpProductsFromDatabase">
    <file:inbound-endpoint path="./data"/>
    <byte-array-to-string-transformer/>
    <!--注释2  分发到JDBC 和 FTP -->
    <all>
        <processor-chain>
        <!--注释3 转换CSV数据为map组成的list -->
        <custom-transformer name="csvToMapTransformer"
                       class="com.prancingdonkey.transformer.CSVToListOfMapsTransformer"/>
        <!--注释4 遍历list的元素-->
            <foreach>
                 <!--注释5 插入到数据库-->
                <jdbc:outbound-endpoint queryKey="productInsert"
                               connector-ref="jdbcConnector"/>
            </foreach>
        </processor-chain>
       <ftp:outbound-endpoint  user="admin"
                           password="123456"
                               host="${ftp.host}"
                               port="${ftp.port}"
                               path="/data/prancingdonkey/catalog"/>
    </all>
</flow>

 

注释1处 定义插入语句。这里需要一个map类型的payload，使用了MEL从payload中取值。

注释2处 开始使用一个 all 路由器，分发产品数据到JDBC outbound endpoint和 FTP endpoint。

注释3处 产品数据是CSV格式的，需要使用一个转换器转为为一个元素为map的list。

注释4处 使用 foreach消息处理器遍历list的每个元素，并调用JDBC outbound endpoint把每个

map insert到数据库。

 

   批量插入  mule的企业版允许执行批量数据库操作。如果你使用的是企业版的JDBC传输，你可以

   容易的把由map组成的list传送给JDBC endpoint。整个list将一次性批量插入数据库，并会避免

   使用foreach处理器 。这也允许你在一个事务中执行插入。

   

 3.7.3 NoSQL的 MongoDB

 

     MongoDB是一个领先的NoSQL数据库。它把数据暴露成JSON文档的集合形式，并使用JSON进行查询。

 MongoDB 在大数据和云计算基础架构之上可以很容易地实现横向扩展。除了其作为一个文件存储设备，

 MongoDB通过GridFS提供用于分布式文件存储方案。本节 学习mule的MongoDB连接器。你将看到如何

 插入文档到集合，以及如何通过HTTP执行查询。

 

 来实现另一个对 cooling.alerts topic 的订阅者。下面的列表展示了 Prancing Donkey如何把制冷

 系统警告存储到MongonDB的集合。（图示见3.19）

Listing 3.24 Save cooling alerts to a MongoDB collection 

<!--配置MongoDB传输 连接MongoDB-->
<mongo:config name="mongoDB"
          database="prancingdonkey"
          username="${mongo.user}"
          password="${mongo.password}"
              host="${mongo.host}"/>
<jms:activemq-connector name="jmsConnector"
               specification="1.1"
                   brokerURL="${jms.url}"/>
<flow name="saveCoolingAlerts">
     <jms:inbound-endpoint topic="cooling.alerts" connector-ref="jmsConnector" />
 <!--转换警告-->
     <mongo:json-to-dbobject/>
 <!--把警告保存到 cooling_alerts 集合-->
     <mongo:insert-object collection="cooling_alerts" config-ref="mongoDB" />
</flow>

 

 

你可能注意到了MongoDB和本章之前见到的 inbound/outbound endpoint 配置有很大的不同。

 这是因为在mule中对MongoDB的支持被实现为云连接器。它们暴露API的单独的方法而不是暴露消息渠道

 来向流获取和发送数据。

 

     因为MongoDB 云连接器暴露了MongoDB java驱动的方法，你需要保证你的消息的payload是API需要的格式。 本例中，你需要在调用 insert-object消息处理器之前使用json-to-dbobject转换器把JSON格式的payload 转为com.mongodb.DBObject的实例的格式。现在来看如何在MongoDB集合中查询文档。（见图3.20）

 

 

 下面的列表定义了一个流，这个流允许用户在一个URL中提供一个JSON查询，以从配置的MongoDB集合中

 获取匹配的文档。（见图3.20）

 Listing 3.25 Query a MongoDB collection over HTTP

 

<flow name="queryCoolingAlerts" >
    <!--注释1 定义http:inbound endpoint 接受查询-->
    <http:inbound-endpoint host="${http.host}"
                           port="${http.port}"
                           path="alerts/cooling"/>
    <byte-array-to-string-transformer/>
    <!--对这个集合执行查询-->
    <mongo:find-objects config-ref="mongoDB"
                        collection="cooling_alerts" />
    <!--转换集合为JSON -->
    <mongo:mongo-collection-to-json/>
</flow>

 

 

注释1处 HTTP inbound endpoint允许一个客户端提交查询以从cooling_alerts 集合返回一个JSON格式的文档set。mule的 MongoDB模块还有其他的多种操作。完整的列表可以查看 www.mulesoft.org/muleforge/mongodb-connector.

 

本节你看到了在mule中如何使用关系型的和NoSQL的数据库。我们演示了如何将JDBC传输作为消息源使用，以从数据库查询反复的生成mule消息。你也看到如何使用JDBC outbound endpoint 向表中插入数据。最后，我们看了如何从MongoDB查询和插入文档。