高负载场景下用Spring JDBC Template高效批量检查DB数据存在性

针对你这个高负载场景，核心是避免逐个查询存在性，而是通过批量查询+批量操作减少数据库交互次数，最大化利用Kafka批量消息的优势。下面是具体实现思路和代码示例：

1. 批量查询已存在的(id, serial_num)组合 #

利用主流数据库的多列IN查询能力（PostgreSQL、MySQL 8.0+等均支持），一次性查出所有待处理消息中已存在于表A的唯一键组合，彻底避免循环查询。

代码实现（Spring JdbcTemplate） #

// 假设实体类为EntityA，包含id、serialNum、isActual字段
private Set<String> queryExistingCombines(List<EntityA> entities) {
    StringBuilder sqlBuilder = new StringBuilder("SELECT CONCAT(id, '|', serial_num) FROM A WHERE (id, serial_num) IN (");
    List<Object[]> params = new ArrayList<>(entities.size());

    // 动态拼接SQL参数
    for (int i = 0; i < entities.size(); i++) {
        if (i > 0) {
            sqlBuilder.append(", ");
        }
        sqlBuilder.append("(?, ?)");
        EntityA entity = entities.get(i);
        params.add(new Object[]{entity.getId(), entity.getSerialNum()});
    }
    sqlBuilder.append(")");

    // 执行查询并返回已存在的组合（用字符串拼接作为唯一标识）
    return jdbcTemplate.query(
        sqlBuilder.toString(),
        params.toArray(new Object[0]),
        (resultSet, rowNum) -> resultSet.getString(1)
    );
}

2. 拆分待处理实体，批量执行更新与插入 #

根据查询结果，把消息分为两类：已存在的直接跳过；不存在的先将对应id下的所有实体is_actual设为false，再批量插入新实体。

核心处理逻辑 #

@Transactional // 事务控制保证操作原子性，避免部分成功部分失败
public void processKafkaBatch(List<EntityA> entities) {
    // 第一步：批量查询已存在的唯一键组合
    Set<String> existingCombines = queryExistingCombines(entities);

    // 第二步：拆分出需要插入的实体和需要失效的id集合
    List<EntityA> toInsert = new ArrayList<>();
    Set<String> idsToDeactivate = new HashSet<>();

    for (EntityA entity : entities) {
        String combineKey = entity.getId() + "|" + entity.getSerialNum();
        if (!existingCombines.contains(combineKey)) {
            toInsert.add(entity);
            idsToDeactivate.add(entity.getId());
        }
    }

    // 第三步：批量失效对应id下的所有旧实体
    if (!idsToDeactivate.isEmpty()) {
        String updateSql = "UPDATE A SET is_actual = false WHERE id IN (:ids)";
        // 用NamedParameterJdbcTemplate更优雅处理IN集合参数
        namedParameterJdbcTemplate.update(
            updateSql,
            Collections.singletonMap("ids", idsToDeactivate)
        );
    }

    // 第四步：批量插入新实体
    if (!toInsert.isEmpty()) {
        String insertSql = "INSERT INTO A (id, serial_num, is_actual) VALUES (?, ?, ?)";
        jdbcTemplate.batchUpdate(insertSql, new BatchPreparedStatementSetter() {
            @Override
            public void setValues(PreparedStatement ps, int index) throws SQLException {
                EntityA entity = toInsert.get(index);
                ps.setString(1, entity.getId());
                ps.setString(2, entity.getSerialNum());
                ps.setBoolean(3, true); // 新插入的实体设为有效状态
            }

            @Override
            public int getBatchSize() {
                return toInsert.size();
            }
        });
    }
}

3. 兼容性与优化建议 #

• 低版本数据库适配：如果你的数据库不支持多列IN查询（比如MySQL 5.x），可以改用临时表：先批量插入待检查的(id, serial_num)到临时表，再通过关联查询找出已存在的组合。
• 性能调优：
  • 确保(id, serial_num)的唯一键索引已创建，加速存在性查询。
  • 根据数据库性能调整批量大小（500条在多数场景下合理，若数据库压力大可拆分为100-200条的子批量）。
  • 开启数据库的批量提交优化（比如MySQL的rewriteBatchedStatements=true，PostgreSQL的useServerPrepStmts=true）。
• 简化逻辑的可选方案：如果业务允许，可结合数据库冲突处理语法（比如PostgreSQL的INSERT ... ON CONFLICT DO NOTHING，MySQL的INSERT ... ON DUPLICATE KEY UPDATE），但注意这种方式无法自动将同id下的旧实体设为失效，仍需额外批量更新步骤。

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