如何将int-jdbc:inbound-channel-adapter查询时间传入更新语句

问题背景 #

我使用Spring Integration的int-jdbc:inbound-channel-adapter从messages表消费消息，通过offsets表记录消费的时间偏移量：

• 查询语句会获取当前时间NOW()作为now_，筛选出creation_date晚于上次消费时间的消息
• 更新语句用REPLACE INTO将本次查询的时间写入offsets表，作为下次消费的基准

遇到的问题：

1. 直接使用原查询时，:now_会被解析为集合（因为查询返回多条记录，每条都带now_），导致更新失败
2. 用静态方法生成时间戳，无法保证和查询时的NOW()完全一致，存在消息遗漏风险
3. 设置update-per-row="true"并改用UPDATE语句，功能正常，但3000条数据处理时间从5分钟增至1.5分钟，性能大幅下降

最优解决方案 #

核心思路是让查询只生成单次时间戳，再通过自定义参数工厂提取这个单一值，既保证时间一致性，又保留批量更新的高性能。

1. 修改查询语句，生成单一基准时间 #

通过子查询提前获取本次查询的时间戳，再关联消息表，确保所有结果记录携带同一个now_值：

SELECT t.now_, m.id, m.payload, m.priority
FROM (SELECT NOW() as now_) t
CROSS JOIN myDb.messages m
WHERE m.creation_date > (
    SELECT o.latest_consultation 
    FROM myDb.offsets o 
    WHERE o.group_id = 'someGroupId'
)

2. 自定义SqlParameterSourceFactory提取单一时间戳 #

实现自定义工厂类，从查询结果的第一条记录中提取now_，避免将所有记录的now_作为集合传入更新语句：

public class SingleValueSqlParameterSourceFactory implements SqlParameterSourceFactory {
    @Override
    public SqlParameterSource createParameterSource(Object input) {
        if (input instanceof List<?>) {
            List<?> resultList = (List<?>) input;
            if (!resultList.isEmpty()) {
                // 适配查询结果的类型：如果是实体类就用getter，这里以Map为例
                Map<?, ?> firstRecord = (Map<?, ?>) resultList.get(0);
                return new MapSqlParameterSource("now_", firstRecord.get("now_"));
            }
        }
        return new MapSqlParameterSource();
    }
}

3. 配置适配器使用自定义工厂 #

保持update-per-row="false"（默认值），使用原REPLACE INTO语句，同时指定自定义参数工厂：

<int-jdbc:inbound-channel-adapter
    channel="messageChannel"
    data-source="dataSource"
    query="SELECT t.now_, m.id, m.payload, m.priority FROM (SELECT NOW() as now_) t CROSS JOIN myDb.messages m WHERE m.creation_date > (SELECT o.latest_consultation FROM myDb.offsets o WHERE o.group_id = 'someGroupId')"
    update="REPLACE INTO myDb.offsets (group_id,latest_consultation) values ('someGroupId', :now_)"
    update-sql-parameter-source-factory="singleValueSqlParameterSourceFactory">
    <int:poller fixed-rate="5000"/>
</int-jdbc:inbound-channel-adapter>

<bean id="singleValueSqlParameterSourceFactory" class="com.example.SingleValueSqlParameterSourceFactory"/>

方案优势 #

• 时间一致性：now_完全等于查询执行时的NOW()值，不会出现消息遗漏
• 高性能：仍然只执行一次REPLACE语句，保持原有的批量处理性能，避免update-per-row带来的开销

另外，如果允许修改数据库逻辑，也可以用存储过程封装“获取时间→查询消息→更新偏移量”的完整逻辑，适配器只需调用存储过程，同样能达到效果，但自定义参数工厂的方案更轻量，无需改动数据库层。

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