与手动创建适配器相比，sourcePollingChannelAdapterSpec 中似乎忽略了 sendTimeout

当我创建IntegrationFlow设置时sendTimeout()，值SourcePollingChannelAdapterSpec被忽略，导致轮询线程阻塞。如果我以编程方式创建，则不会SourcePollingChannelAdapter发生这种情况。例如，以编程方式使用@Configuration类：

@Configuration
public class JdbcPollingConfig {

    @Bean
    protected MessageChannel jdbcInboundChannel() {
        return new QueueChannel(1);
    }

    @Bean
    public ThreadPoolTaskScheduler jdbcTaskExecutor() {
        ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
        taskScheduler.setPoolSize(1);
        taskScheduler.initialize();
        return taskScheduler;
    }

    @Bean
    protected JdbcPollingChannelAdapter jdbcPollingChannelAdapter(final DataSource dataSource) {
        final JdbcPollingChannelAdapter adapter =
                new CustomJdbcPollingChannelAdapter(dataSource, "select dp_id, dp_payload from data_packets");
        return adapter;
    }

    @Bean
    protected SourcePollingChannelAdapter sourcePollingChannelAdapter(
            final JdbcPollingChannelAdapter adapter,
            final MessageChannel jdbcInboundChannel,
            final ThreadPoolTaskScheduler jdbcTaskExecutor) {
        final SourcePollingChannelAdapter spcAdapter = new SourcePollingChannelAdapter();
        spcAdapter.setSource(adapter);
        spcAdapter.setOutputChannel(jdbcInboundChannel);
        spcAdapter.setSendTimeout(500); // Sets the send-to-channel timeout - throws exception on timeout though.
        spcAdapter.setTaskScheduler(jdbcTaskExecutor);
        spcAdapter.setTrigger( new PeriodicTrigger(Duration.ofSeconds(2L)));
        return spcAdapter;
    }

}

请注意，该类CustomJdbcPollingChannelAdapter是我自己的，它只是进行了扩展，以便我可以doReceive()使用一些日志调用来覆盖该方法：

@Log4j2
public class CustomJdbcPollingChannelAdapter extends JdbcPollingChannelAdapter {
    public CustomJdbcPollingChannelAdapter(DataSource dataSource, String selectQuery) {
        super(dataSource, selectQuery);
    }

    public CustomJdbcPollingChannelAdapter(JdbcOperations jdbcOperations, String selectQuery) {
        super(jdbcOperations, selectQuery);
    }

    @Override
    protected Object doReceive() {
        log.info("Jdbc Polling.");
        return super.doReceive();
    }
}

在配置了通道队列的情况下，此配置会按预期抛出异常，因为没有通道的消费者jdbcInboundChannel。对比一下使用和IntegrationFlow配置：

@Configuration
public class JdbcDSLConfig {

    @Bean
    public QueueChannelSpec jdbcInboundChannel() {
        return MessageChannels.queue(1);
    }

    @Bean
    public MessageSource<Object> jdbcMessageSource(final DataSource dataSource) {
        return new CustomJdbcPollingChannelAdapter(dataSource, "select dp_id, dp_payload from data_packets");
    }

    @Bean
    public ThreadPoolTaskExecutor jdbcExecutor() {
        final ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        taskExecutor.setCorePoolSize(1);
        taskExecutor.setMaxPoolSize(1);
        taskExecutor.initialize();
        return taskExecutor;
    }

    @Bean
    public IntegrationFlow jdbcInboundFlow(final MessageSource<Object> jdbcMessageSource,
                                           final QueueChannelSpec jdbcInboundChannel,
                                           @Qualifier("jdbcExecutor") final ThreadPoolTaskExecutor jdbcExecutor) {
        return IntegrationFlow.from(jdbcMessageSource,
                        c -> c.poller(Pollers
                                .fixedDelay(2000)
                                .sendTimeout(1) // this has no effect.
                                .taskExecutor(jdbcExecutor)))
                .channel(jdbcInboundChannel)
                .get();

    }
}

即使在配置中设置了 sendTimeout，此配置也会在数据库的第二次轮询时阻塞。