当前位置：首页 > news >正文

Hash查询方案对比分析

news 2026/5/14 21:54:53

推荐方案：`MD5(concat_ws)` 显式标准化

核心 SQL 模板（动态生成）：

sql

复制

-- 通用查询模板（PgSQL 和 GaussDB 两侧均适用）
SELECTid,MD5(concat_ws('|',COALESCE(col_int::text,       'NULL'),COALESCE(col_numeric::text,   'NULL'),COALESCE(col_ts::text,        'NULL'),  -- 注意时区处理见下方COALESCE(col_bool::text,      'NULL'),COALESCE(col_text,            'NULL'))) AS row_hash
FROM your_table
WHERE date_col >= '2026-01-01' AND date_col < '2026-02-01'
ORDER BY id;

各数据类型的标准化处理细节

这是两库差异的重灾区，必须逐类型处理：

数据类型	问题	推荐处理方式
`numeric` / `decimal`	`1.5` vs `1.50` 精度不同	`to_char(col, 'FM999999999990.##########')` 或 `CAST(col AS numeric(p,s))::text`
`timestamp`	时区偏移字符串格式不同	统一转 UTC：`to_char(col AT TIME ZONE 'UTC', 'YYYY-MM-DD HH24:MI:SS.US')`
`timestamptz`	同上 + 时区标记差异	`to_char(col AT TIME ZONE 'UTC', 'YYYY-MM-DD HH24:MI:SS.US')`
`boolean`	`t`/`f` vs `true`/`false`	`CASE WHEN col THEN '1' ELSE '0' END`
`float` / `double`	浮点精度表示不同	`to_char(col, 'FM0.####################')` 或约定精度位数
`jsonb`	键序不保证一致	两侧都用 `jsonb_strip_nulls(col)::text`（或应用层排序后 hash）
`array`	格式差异	`array_to_string(col, ',')`
`NULL`	不处理会被 concat 忽略	必须 `COALESCE(col::text, 'NULL')`

问题二：千万行级别的高效比对方案

整体架构设计

具体实现策略

策略一：分片 + 线程池（核心）

java

复制

// 推荐技术栈（Spring Boot + CompletableFuture）
@Service
public class DataCompareService {private final ExecutorService compareExecutor = new ThreadPoolExecutor(8, 16,              // coreSize=8, maxSize=1660L, TimeUnit.SECONDS,new LinkedBlockingQueue<>(200),new ThreadFactory(/* 命名 compare-worker-N */),new CallerRunsPolicy()  // 背压：队满时调用方线程自己执行，避免OOM);public CompareReport compare(CompareTask task) {// 1. 先查总行数，决定分片数long total = pgRepo.countByDateRange(task.getTable(), task.getDateRange());int shardSize = 200_000; // 每片 20w 行int shardCount = (int) Math.ceil((double) total / shardSize);// 2. 生成分片任务，并发提交List<CompletableFuture<ShardResult>> futures = new ArrayList<>();for (int i = 0; i < shardCount; i++) {final long offset = (long) i * shardSize;futures.add(CompletableFuture.supplyAsync(() -> compareOneShard(task, offset, shardSize),compareExecutor));}// 3. 汇聚结果return CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).thenApply(v -> futures.stream().map(CompletableFuture::join).collect(aggregatingCollector())).join();}
}

策略二：每个 Worker 内部——同一分片对两库并发查询

java

复制

private ShardResult compareOneShard(CompareTask task, long offset, int limit) {// PgSQL 和 Gauss 的查询并发执行（不是串行！）CompletableFuture<Map<Long, String>> pgFuture = CompletableFuture.supplyAsync(() -> pgHashQuery(task, offset, limit), ioExecutor);CompletableFuture<Map<Long, String>> gaussFuture = CompletableFuture.supplyAsync(() -> gaussHashQuery(task, offset, limit), ioExecutor);Map<Long, String> pgHashes   = pgFuture.join();    // Map<id, hash>Map<Long, String> gaussHashes = gaussFuture.join();return diffMaps(pgHashes, gaussHashes);
}

关键点：每个分片内，PgSQL 和 GaussDB 的查询是并发的，等待时间取决于慢的那个。

策略三：流式读取（防止 OOM）

当单片数据量仍然较大时，使用游标/流式模式而非一次性 fetchAll：

java

复制

// Spring JDBC fetchSize 设置（关闭自动提交，启用游标）
@Transactional(readOnly = true)
public Map<Long, String> pgHashQueryStream(CompareTask task, long offset, int limit) {Map<Long, String> result = new HashMap<>(limit, 0.75f);jdbcTemplate.setFetchSize(1000);  // 每次从数据库获取 1000 行jdbcTemplate.query(buildHashSql(task, offset, limit), rs -> {result.put(rs.getLong("id"), rs.getString("row_hash"));});return result;
}

策略四：两侧结果的高效 diff

java

复制

private ShardResult diffMaps(Map<Long, String> pgMap, Map<Long, String> gaussMap) {List<Long> onlyInPg    = new ArrayList<>();  // Gauss 缺失List<Long> onlyInGauss = new ArrayList<>();  // PgSQL 缺失List<Long> hashDiff    = new ArrayList<>();  // 两边都有但 hash 不同// PgSQL 视角遍历pgMap.forEach((id, pgHash) -> {String gaussHash = gaussMap.get(id);if (gaussHash == null)          onlyInPg.add(id);else if (!pgHash.equals(gaussHash)) hashDiff.add(id);});// Gauss 独有gaussMap.forEach((id, _) -> {if (!pgMap.containsKey(id)) onlyInGauss.add(id);});return new ShardResult(onlyInPg, onlyInGauss, hashDiff);
}

线程池设计建议

参数	推荐值	说明
分片大小	10w~50w 行	过小则线程切换开销大；过大则单片慢
并发线程数	数据库连接池上限的 1/2	避免耗尽连接池，留给其他业务
队列类型	`LinkedBlockingQueue(200)`	有界队列，防止任务堆积 OOM
拒绝策略	`CallerRunsPolicy`	背压，不丢任务
`fetchSize`	1000~5000	启用游标，防大结果集 OOM

进阶优化（可选）

分两阶段比对（先粗后细）：

第 1 阶段：按分片聚合 XOR Hash → 秒级发现哪些分片有差异（无需全量 join）SELECT MIN(id), MAX(id), BIT_XOR(hashtext(row_hash)::bigint) FROM ...↓ 仅对 XOR 不一致的分片
第 2 阶段：精确行级比对 → 只对有差异的分片做完整 id+hash 比对

这在"大多数数据一致"的场景下能节省 80%+ 的网络和计算开销。

总结

问题	推荐方案
Hash 方式	`MD5(concat_ws('
NULL 处理	必须 `COALESCE(col::text, 'NULL')`
时间类型	统一 `to_char(col AT TIME ZONE 'UTC', ...)`
千万级性能	分片（10w~50w/片）+ 线程池 + 单片内 PgSQL/Gauss 并发查询
内存控制	`fetchSize=1000`，游标流式读取
进一步加速	先 XOR 粗筛分片，再精确行级比对