spring-integration - 如何限制 JdbcPollingChannelAdapter 的事务边界

问题描述

我有一个 JdbcPollingChannelAdapter 定义如下：

@Bean
public MessageSource<Object> jdbcMessageSource(DataSource dataSource) {
    JdbcPollingChannelAdapter jdbcPollingChannelAdapter = new JdbcPollingChannelAdapter(dataSource,
            "SELECT * FROM common_task where due_at <= NOW() and retries < order by due_at ASC FOR UPDATE SKIP LOCKED");
    jdbcPollingChannelAdapter.setMaxRowsPerPoll(1);
    jdbcPollingChannelAdapter.setUpdateSql("Update common_task set retries = :retries, due_at = due_at + interval '10 minutes' WHERE ID = (:id)");
    jdbcPollingChannelAdapter.setUpdatePerRow(true);
    jdbcPollingChannelAdapter.setRowMapper(this::mapRow);
    jdbcPollingChannelAdapter.setUpdateSqlParameterSourceFactory(this::updateParamSource);
    return jdbcPollingChannelAdapter;
}

集成流程：

@Bean
public IntegrationFlow pollingFlow(MessageSource<Object> jdbcMessageSource) {
    return IntegrationFlows.from(jdbcMessageSource,
            c -> c.poller(Pollers.fixedRate(250, TimeUnit.MILLISECONDS)
                    .maxMessagesPerPoll(1)
                    .transactional()))
            .split()
            .channel(taskSourceChannel())
            .get();
}

服务激活器定义为

@ServiceActivator(inputChannel = "taskSourceChannel")
    public void doSomething(FooTask event) {
        //do something but ** not ** within the transaction of the poller.
    }

集成流中的轮询器被定义为事务性的。根据我的理解，这将 1. 在事务中执行选择查询和更新查询。2. 它还会在同一个事务中执行 doSomething() 方法。

目标：我想做 1 而不是 2。我想在事务中进行选择和更新，以确保两者都发生。但是，我不想在同一个事务中执行 doSomething()。如果 doSomething() 中出现异常，我仍然希望保留在轮询期间所做的更新。我怎样才能做到这一点？

标签： spring-integrationspring-integration-dsl

这是通过简单的线程移动来完成的。所以，你需要的只是离开轮询线程，让它提交 TX 并在单独的线程中继续处理。

根据您的逻辑.split()，最好在拆分后已经有新的线程处理，因此项目甚至会被doSomething()并行处理。

只需使用ExecutorChannel. 既然您已经拥有了taskSourceChannel()，只需考虑将其替换为 ExecutorChannel基于一些托管的ThreadPoolTaskExecutor.

以及它的 Javadocs。

简单的 Java 配置变体是这样的：

    @Bean
    public MessageChannel taskSourceChannel() {
        return new ExecutorChannel(executor());
    }

    @Bean
    public Executor executor() {
        return new ThreadPoolTaskExecutor();
    }

spring-integration - 如何限制 JdbcPollingChannelAdapter 的事务边界

问题描述

解决方案

推荐阅读