如何编程实现JDBC入站通道适配器读取库表数据并流转至业务处理？

以下是你提供的XML配置对应的全流程编程实现，完全匹配原XML的消息流转逻辑：

核心配置类实现 #

@Configuration
@EnableIntegration
public class JdbcIntegrationConfig {

    @Autowired
    private DataSource dataSource;

    @Autowired
    private AccountRowMapper accountRowMapper;

    @Autowired
    private AccountService accountService;

    // 定义任务执行器，对应XML中的executorPool
    @Bean
    public TaskExecutor executorPool() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(5);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(20);
        executor.initialize();
        return executor;
    }

    // 定义OutChannel，对应XML中的<int:channel id="OutChannel"/>
    @Bean
    public MessageChannel outChannel() {
        return new DirectChannel();
    }

    // 定义带任务执行器的AccountOutputChannel，对应XML中的<int:channel id="AccountOutputChannel"/>
    @Bean
    public MessageChannel accountOutputChannel() {
        return new ExecutorChannel(executorPool());
    }

    // 构建完整消息流转流程
    @Bean
    public IntegrationFlow jdbcInboundFlow() {
        return IntegrationFlows.from(
                        // 配置JDBC入站适配器，对应XML的<int-jdbc:inbound-channel-adapter>
                        Jdbc.inboundAdapter(dataSource)
                                .sql("SELECT * FROM Accounts")
                                .rowMapper(accountRowMapper)
                                .maxRowsPerPoll(100),
                        // 配置轮询器，对应XML的<int:poller fixed-rate="1000"/>
                        e -> e.poller(Pollers.fixedRate(1000)))
                // 将查询结果发送至OutChannel
                .channel(outChannel())
                // 桥接器，转发消息至AccountOutputChannel，对应XML的<int:bridge>
                .bridge(e -> e.outputChannel(accountOutputChannel()))
                // 服务激活器，调用AccountService的onMessage方法处理消息，对应XML的<int:service-activator>
                .handle(accountService, "onMessage")
                .get();
    }
}

关键组件对应说明 #

• 任务执行器：executorPool() 实现了XML中<int:dispatcher task-executor="executorPool"/>的线程池功能，可根据业务需求调整线程池参数。
• 通道定义：
  • outChannel() 是直接通道，和XML中<int:channel id="OutChannel"/>的无分发器通道逻辑一致。
  • accountOutputChannel() 是ExecutorChannel，绑定任务执行器，实现异步消息分发。
• IntegrationFlow流程：
  • Jdbc.inboundAdapter() 完全复刻XML中JDBC入站适配器的配置，包括SQL查询、行映射、轮询行数限制。
  • poller(Pollers.fixedRate(1000)) 配置1秒一次的轮询频率，和XML的轮询器配置一致。
  • .bridge() 方法实现桥接器的消息转发功能，连接OutChannel和AccountOutputChannel。
  • .handle() 方法绑定AccountService的onMessage方法，作为消息的最终处理器。

注意事项 #

确保项目中已引入Spring Integration JDBC依赖（如Maven的spring-integration-jdbc），同时Spring Context、Spring JDBC等基础依赖配置完整。

内容的提问来源于stack exchange，提问作者sammy