Spring Integration：无需自定义实现解决JDBC入站通道适配器并发重复处理

我开发了一个基于Spring Integration的微服务，集成流以JDBC入站通道适配器为起点。使用该适配器时，可配置更新查询将已读取的数据行标记为已处理，避免重复处理。但当运行该微服务的多个副本时，两个实例的适配器可能同时查询数据库，导致同一份结果集被两个实例获取，进而引发重复处理问题。我目前通过自定义数据源，结合事务与select for update查询实现了并发读取同步，请问是否可以无需此类自定义实现，直接通过JDBC入站通道适配器本身实现该功能？

当然可以，Spring Integration的JDBC入站通道适配器本身就支持通过悲观锁或行级锁机制来避免多实例下的重复读取，无需自定义数据源。具体可以通过以下两种方式实现：

方式一：使用select ... for update skip locked（推荐） #

直接在适配器的查询语句中使用数据库支持的skip locked语法（如PostgreSQL、MySQL 8.0+等均支持），多个实例同时查询时，已被其他事务锁定的行会被跳过，每个实例仅能获取未被锁定的行。配合适配器的更新查询标记已处理状态，就能彻底避免重复处理。

示例XML配置：

<int-jdbc:inbound-channel-adapter
    query="SELECT id, data FROM tasks WHERE status = 'PENDING' FOR UPDATE SKIP LOCKED"
    update="UPDATE tasks SET status = 'PROCESSING' WHERE id IN (:id)"
    channel="inputChannel"
    data-source="dataSource"
    update-per-row="false">
    <int:poller fixed-rate="1000"/>
</int-jdbc:inbound-channel-adapter>

示例Java配置：

@Bean
public JdbcPollingChannelAdapter jdbcPollingChannelAdapter(DataSource dataSource) {
    JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(dataSource,
            "SELECT id, data FROM tasks WHERE status = 'PENDING' FOR UPDATE SKIP LOCKED");
    adapter.setUpdateSql("UPDATE tasks SET status = 'PROCESSING' WHERE id IN (:id)");
    adapter.setUpdatePerRow(false);
    return adapter;
}

方式二：利用适配器的事务与锁机制 #

JDBC入站通道适配器默认会在事务中执行查询和更新操作，你可以配置transaction-manager确保整个查询-更新流程在同一个事务中，同时将查询语句改为select ... for update。第一个实例查询到行后会锁定这些行，其他实例在事务提交前无法读取到这些行；直到第一个实例完成更新并提交事务，行的状态被修改后，其他实例就不会再读取到已处理的行。

这种方式需要注意设置合理的事务超时时间，避免锁持有过久影响系统性能。同时要确保查询的过滤条件（如status = 'PENDING'）有对应的索引，避免全表扫描导致锁表。

总结来说，无需自定义数据源，只需调整适配器的查询语句并结合事务配置，就能实现多实例下的并发读取同步，避免重复处理。

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