如何使用 Mule 在时间间隔内从数据库中获取特定数量的行答案

【问题标题】：How to fetch particular numbers of rows from DB in a time interval using Mule如何使用 Mule 在时间间隔内从数据库中获取特定数量的行
【发布时间】：2017-04-10 03:39:56
【问题描述】：

我有一个 Mule 流程，我需要从数据库中获取行并写入文件。现在我在数据库中有 100 行，我需要一次从数据库中获取 5 行并再次写入文件间隔一段时间后说 30 秒再获取 5 行并将有效负载写入文件.. 现在我的流程如下：-

 <spring:beans>
        <spring:bean id="DB_Source" name="DB_Source" class="org.enhydra.jdbc.standard.StandardDataSource">
            <spring:property name="url" value="${url}"/>
            <spring:property name="driverName" value="${driverName}"/>
        </spring:bean>
     </spring:beans>
    <jdbc-ee:connector name="Database_Global" dataSource-ref="DB_Source" validateConnections="true" queryTimeout="-1" pollingFrequency="0" doc:name="Database" transactionPerMessage="true">
        <!-- Here transactionPerMessage="false" so that it retrieve and display all the row at once-->
         <jdbc-ee:query key="RetriveQuery" value="select * from getData"/>  <!-- or we can use CALL sp_retrieveData(@Id=13) -->
    </jdbc-ee:connector>
    <context:property-placeholder location="classpath:conf/DBConnectionProp.properties"/>



    <flow name="InboundJDBC" doc:name="InboundJDBC" initialState="started">
        <jdbc-ee:inbound-endpoint  queryTimeout="-1" pollingFrequency="1000" doc:name="Database"   connector-ref="Database_Global" queryKey="RetriveQuery">

         <jdbc-ee:transaction action="ALWAYS_BEGIN" />

        <!--  <property key="receiveMessageInTransaction" value="true"/> --><!-- This to receive all the row in once -->
        </jdbc-ee:inbound-endpoint>
        <mulexml:object-to-xml-transformer doc:name="Object to XML"/>

      <message-properties-transformer doc:name="Message Properties"> 
      <add-message-property key="MULE_CORRELATION_GROUP_SIZE" value="5"/> <!-- Set the number of rows to be return at a time -->
      <add-message-property key="MULE_CORRELATION_ID" value="1"/> 
      </message-properties-transformer> 
      <collection-aggregator timeout="5000" failOnTimeout="false" doc:name="Collection Aggregator"/>

        <logger message="JDBC Transaction #[message.payload] **************" level="INFO" doc:name="Logger"/>
        <file:outbound-endpoint path="E:\backup\test\ss" outputPattern="#[java.util.UUID.randomUUID().toString()].txt" responseTimeout="10000" doc:name="File"/>

    </flow>  
</mule>

现在的问题是，当应用程序启动时，它只从数据库中提取 100 行中的 5 行并写入文件，然后不会提取剩余的行，也不会创建新文件......但我想要每 30 秒后获取 5 行并在最后将其写入一个新文件.. 我做错什么了吗？我已将以下内容作为参考：- How do I get a Mule to return multiple rows from a JDBC query as a single transaction?

更新流程：-

<flow name="InboundJDBC" doc:name="InboundJDBC" initialState="started">
        <jdbc-ee:inbound-endpoint  queryTimeout="-1" pollingFrequency="1000" doc:name="Database"   connector-ref="Database_Global" queryKey="RetriveQuery">

         <jdbc-ee:transaction action="ALWAYS_BEGIN" />

     <!--  <property key="receiveMessageInTransaction" value="true"/> --><!-- This to receive all the row in once -->
        </jdbc-ee:inbound-endpoint>
        <set-property propertyName="#[message.inboundProperties['requestId']]" value="#[java.util.UUID.randomUUID().toString()]" doc:name="Property"/>

        <mulexml:object-to-xml-transformer doc:name="Object to XML"/>

      <message-properties-transformer doc:name="Message Properties"> 

      <add-message-property key="MULE_CORRELATION_GROUP_SIZE" value="5"/> <!-- Set the number of rows to be return at a time -->
      <add-message-property key="MULE_CORRELATION_ID" value="#[message.inboundProperties['requestId']]"/> 
      </message-properties-transformer> 
      <collection-aggregator timeout="5000" failOnTimeout="false" doc:name="Collection Aggregator"/>

        <logger message="JDBC Transaction #[message.payload] **************" level="INFO" doc:name="Logger"/>
        <file:outbound-endpoint path="E:\backup\test\ss" outputPattern="#[java.util.UUID.randomUUID().toString()].txt" responseTimeout="10000" doc:name="File"/>

    </flow>

现在它正在为每行创建文件...

【问题讨论】：

轮询频率是1000，是1秒，不是你说的30秒。

标签： jdbc mule mule-studio mule-el

【解决方案1】：

我发现了几个问题：

查询选择所有行而不是只选择 5 行，
没有用于标记所选记录的更新查询，因此将一次又一次地提取相同的记录，
相关 ID 固定为：<add-message-property key="MULE_CORRELATION_ID" value="1"/>。由于相关 ID 是固定的，collection-aggregator 将针对此 ID 聚合 5 条消息，并将停在那里。任何针对此 ID 的新消息都将被丢弃。而是使用为五行生成相同值的 MEL 表达式：使用什么表达式取决于您，例如，它可以是某种时间模数，为 30 秒时间窗口提供恒定值...

【讨论】：

嗨大卫，我已经修改了以下内容：- .....现在它正在生成文件对于每一行..我想每5行有一个文件..我没有使用更新查询，因为它可以选择相同的记录..一个文件应该包含5条记录，并且应该在给定的时间间隔后生成..如何做吗？
如果您对每个尺寸使用不同的相关 ID，则不会发生分组。重新阅读我的答案：您需要制作一个 MEL 表达式，每次轮询数据库时都会产生不同的 ID，但对于 5 条记录中的每条记录都保持相同。
嗨大卫，我绑定了在消息属性转换器之前，然后在消息属性转换器中的 .. 但没有任何工作！它只获取 5 行并被停止..你能否建议我如何实现你提到的事情......
有一个问题：value="requestId" 会将值设置为"requestId"。你想要value="#[message.inboundProperties['requestId']]"。
请查看我在问题中更新的流程.. 根据您的评论，我已设置 #[message.inboundProperties['requestId']] .. 但没有用.. 每个文件都写入多个文件包含单行...

【解决方案2】：

因此，根据 David 关于第 3 点的建议，我将 correlation ID 固定为：<add-message-property key="MULE_CORRELATION_ID" value="1"/>。
从现在开始，correlation ID 已修复，集合聚合器聚合此 ID 的所有消息并停止。
希望这可以帮助遇到同样问题的其他人，并为我工作。

【讨论】：