2

我正在尝试使用以 kafka 作为代理的 spring 云数据流从 oracle DB 进行更改数据捕获。我为此使用轮询机制。我定期使用基本选择查询轮询数据库以捕获任何更新的数据。为了更好的防故障系统，我将上次轮询时间保存在 oracle DB 中，并使用它来获取上次轮询后更新的数据。

public MessageSource<Object> jdbcMessageSource() {
    JdbcPollingChannelAdapter jdbcPollingChannelAdapter =
            new JdbcPollingChannelAdapter(this.dataSource, this.properties.getQuery());
    jdbcPollingChannelAdapter.setUpdateSql(this.properties.getUpdate());
    return jdbcPollingChannelAdapter;
}

@Bean
public IntegrationFlow pollingFlow() {
    IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(jdbcMessageSource(),spec -> spec.poller(Pollers.fixedDelay(3000)));
    flowBuilder.channel(this.source.output());
    flowBuilder.transform(trans,"transform");
    return flowBuilder.get();

}

我在应用程序属性中的查询如下：

query: select * from kafka_test where LAST_UPDATE_TIME >(select LAST_POLL_TIME from poll_time)

update : UPDATE poll_time SET LAST_POLL_TIME = CURRENT_TIMESTAMP

这对我来说非常有用。我可以通过这种方法从数据库中获取 CDC。

我现在正在查看的问题如下：

仅仅为了维护轮询时间而创建一个表是一种负担。我正在寻找在 kafka 主题中维护最后一次投票时间，并在我进行下一次投票时从 kafka 主题中检索该时间。

我已经修改了jdbcMessageSource如下方法来尝试：

public MessageSource<Object> jdbcMessageSource() {
    String query = "select * from kafka_test where LAST_UPDATE_TIME > '"+<Last poll time value read from kafka comes here>+"'";

    JdbcPollingChannelAdapter jdbcPollingChannelAdapter =
            new JdbcPollingChannelAdapter(this.dataSource, query);
    return jdbcPollingChannelAdapter;
}

但是 Spring Data Flow 只实例化 pollingFlow()（请参见上面的代码）bean 一次。因此，首先运行的查询将保持不变。我想用每次轮询的新轮询时间来更新查询。

Integrationflow有没有一种方法可以让我在每次进行民意调查时编写一个自定义来更新这个查询？

我已经尝试过IntegrationFlowContext，但没有成功。

提前致谢 ！！！

4

3 回答 3

2

在上述两个答案的帮助下，我能够找出方法。编写 ajdbc template并将其包装为 bean 并将其用于Integration Flow.

@EnableBinding(Source.class)
@AllArgsConstructor
public class StockSource {

  private DataSource dataSource;

  @Autowired
  private JdbcTemplate jdbcTemplate;

  private MessageChannelFactory messageChannelFactory;  // You can use normal message channel which is available in spring cloud data flow as well.

  private List<String> findAll() {
    jdbcTemplate = new JdbcTemplate(dataSource);
    String time = "10/24/60" . (this means 10 seconds for oracle DB)
    String query = << your query here like.. select * from test where (last_updated_time > time) >>;
    return jdbcTemplate.query(query, new RowMapper<String>() {
      @Override
      public String mapRow(ResultSet rs, int rowNum) throws SQLException {
          ...
          ...
          any row mapper operations that you want to do with you result after the poll.
          ...
          ...
          ...
        // Change the time here for the next poll to the DB. 
        return result;
      }
    });
  }

  @Bean
  public IntegrationFlow supplyPollingFlow() {

    IntegrationFlowBuilder flowBuilder = IntegrationFlows
        .from(this::findAll, spec -> {
          spec.poller(Pollers.fixedDelay(5000));
        });
    flowBuilder.channel(<<Your message channel>>);
    return flowBuilder.get();
  }

}

在我们的用例中，我们在 kafka 主题中坚持最后一次轮询时间。这是为了减少应用程序状态。现在对数据库的每一次新轮询，都会有一个新的时间where。

PS：您的消息代理（kafka/rabbit mq）应该在您的本地运行，或者如果托管在不同的平台上，则连接到它们。

神速！！！

于 2019-04-08T13:07:48.567 回答
1

有关标准适配器中动态查询的机制，请参见 Artem 的答案；然而，另一种选择是简单地将 a 包装JdbcTemplate在一个 Bean 中并调用它

IntegrationFlows.from(myPojo(), "runQuery", e -> ...)
    ...

甚至是一个简单的 lambda

    .from(() -> jdbcTemplate...)
于 2019-03-07T14:25:07.700 回答
1

我们有这个测试配置（对不起，它是一个 XML）：

<inbound-channel-adapter query="select * from item where status=:status" channel="target"
                             data-source="dataSource" select-sql-parameter-source="parameterSource"
                             update="delete from item"/>


    <beans:bean id="parameterSource" factory-bean="parameterSourceFactory"
                factory-method="createParameterSourceNoCache">
        <beans:constructor-arg value=""/>
    </beans:bean>

    <beans:bean id="parameterSourceFactory"
                class="org.springframework.integration.jdbc.ExpressionEvaluatingSqlParameterSourceFactory">
        <beans:property name="parameterExpressions">
            <beans:map>
                <beans:entry key="status" value="@statusBean.which()"/>
            </beans:map>
        </beans:property>
        <beans:property name="sqlParameterTypes">
            <beans:map>
                <beans:entry key="status" value="#{ T(java.sql.Types).INTEGER}"/>
            </beans:map>
        </beans:property>
    </beans:bean>

    <beans:bean id="statusBean"
                class="org.springframework.integration.jdbc.config.JdbcPollingChannelAdapterParserTests$Status"/>

注意ExpressionEvaluatingSqlParameterSourceFactory及其createParameterSourceNoCache()工厂。此结果可用于select-sql-parameter-source.

对此事JdbcPollingChannelAdapter有意见。setSelectSqlParameterSource

因此，您将 a 配置ExpressionEvaluatingSqlParameterSourceFactory为能够将某些查询参数解析为某些 bean 方法调用的表达式，以从 Kafka 获取所需的值。然后createParameterSourceNoCache()将帮助您获得一个预期的SqlParameterSource。

文档中也有一些信息：https ://docs.spring.io/spring-integration/docs/current/reference/html/#jdbc-inbound-channel-adapter

于 2019-03-07T14:27:39.257 回答