diff --git a/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/domain/dto/SyncActionEnum.java b/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/domain/dto/SyncActionEnum.java index dbf9b488..04fccf7b 100644 --- a/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/domain/dto/SyncActionEnum.java +++ b/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/domain/dto/SyncActionEnum.java @@ -37,7 +37,9 @@ public enum SyncActionEnum { /** 冲突：删除恢复（非归属方删除被撤销，数据回写） */ CONFLICT_DELETE_RESTORE("CONFLICT_DELETE_RESTORE", "非归属方删除恢复"), /** 类型转换冲突 */ - CONFLICT_TYPE("CONFLICT_TYPE", "类型转换异常"); + CONFLICT_TYPE("CONFLICT_TYPE", "类型转换异常"), + /** 唯一键冲突（非主键）：INSERT 撞到表上其他唯一索引，跳过不重试 */ + UNIQUE_KEY_CONFLICT("UNIQUE_KEY_CONFLICT", "唯一键冲突(跳过)"); private final String code; private final String desc; diff --git a/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/exception/UniqueKeyConflictException.java b/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/exception/UniqueKeyConflictException.java new file mode 100644 index 00000000..55843322 --- /dev/null +++ b/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/exception/UniqueKeyConflictException.java @@ -0,0 +1,34 @@ +package com.zdxt.enforcementcode.businessdatasyn.exception; + +/** + * 唯一键冲突异常（非主键冲突） + *

用于标识 INSERT 时撞到了表上其他唯一索引（如 enterprise_information.unique(enterprise_number)）， + * 而非主键重复。Executor 层识别后记入 biz_sync_log 并跳过，不增加 retry_count。

+ * + * @author zrb + */ +public class UniqueKeyConflictException extends RuntimeException { + + private final String tableName; + private final String pkValue; + private final String conflictDetail; + + public UniqueKeyConflictException(String tableName, String pkValue, String conflictDetail, Throwable cause) { + super(conflictDetail, cause); + this.tableName = tableName; + this.pkValue = pkValue; + this.conflictDetail = conflictDetail; + } + + public String getTableName() { + return tableName; + } + + public String getPkValue() { + return pkValue; + } + + public String getConflictDetail() { + return conflictDetail; + } +} diff --git a/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/executor/TableSyncExecutor.java b/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/executor/TableSyncExecutor.java index a4719ce9..1a95df5e 100644 --- a/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/executor/TableSyncExecutor.java +++ b/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/executor/TableSyncExecutor.java @@ -6,6 +6,7 @@ import com.zdxt.enforcementcode.businessdatasyn.config.SyncTableConfig; import com.zdxt.enforcementcode.businessdatasyn.domain.dto.*; import com.zdxt.enforcementcode.businessdatasyn.domain.entity.BizSyncRecord; import com.zdxt.enforcementcode.businessdatasyn.domain.entity.BizSyncTask; +import com.zdxt.enforcementcode.businessdatasyn.exception.UniqueKeyConflictException; import com.zdxt.enforcementcode.businessdatasyn.reader.V1SourceReader; import com.zdxt.enforcementcode.businessdatasyn.reader.V2SourceReader; import com.zdxt.enforcementcode.businessdatasyn.resolver.OwnerSideResolver; @@ -162,6 +163,14 @@ public class TableSyncExecutor { true, null, (int)(System.currentTimeMillis() - startMs)); } + } catch (UniqueKeyConflictException e) { + // 唯一键冲突：不增加 retry_count（重试也没用），只记日志，交人工裁决 + totalSkip++; + log.warn("[BizSync] 表 {} 行 pk={} 唯一键冲突,跳过(不重试): {}", + configKey, pk, e.getConflictDetail()); + syncLogService.log(currentTask.getTaskId(), configKey, pk, + SyncActionEnum.UNIQUE_KEY_CONFLICT.getCode(), null, null, null, null, null, + true, e.getConflictDetail(), (int)(System.currentTimeMillis() - startMs)); } catch (Exception e) { totalFail++; log.error("[BizSync] 表 {} 行 pk={} 同步失败: {}", configKey, pk, e.getMessage(), e); diff --git a/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/writer/V1SinkWriter.java b/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/writer/V1SinkWriter.java index ba07e3df..8ea2677e 100644 --- a/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/writer/V1SinkWriter.java +++ b/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/writer/V1SinkWriter.java @@ -5,8 +5,10 @@ import com.zdxt.enforcementcode.businessdatasyn.config.SyncLogicalDeleteConfig; import com.zdxt.enforcementcode.businessdatasyn.config.SyncTableConfig; import com.zdxt.enforcementcode.businessdatasyn.config.SyncTableSideConfig; import com.zdxt.enforcementcode.businessdatasyn.domain.dto.RowSnapshot; +import com.zdxt.enforcementcode.businessdatasyn.exception.UniqueKeyConflictException; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.dao.DuplicateKeyException; import org.springframework.jdbc.core.namedparam.MapSqlParameterSource; import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate; import org.springframework.stereotype.Service; @@ -67,31 +69,71 @@ public class V1SinkWriter implements SinkWriter { throw new IllegalStateException("Upsert 没有有效列: " + side.getTable()); } - // INSERT INTO table (col1, col2, ...) VALUES (:col1, :col2, ...) - // ON DUPLICATE KEY UPDATE col1 = VALUES(col1), col2 = VALUES(col2), ... - StringJoiner colJoiner = new StringJoiner(", "); - StringJoiner valJoiner = new StringJoiner(", "); - StringJoiner updateJoiner = new StringJoiner(", "); + String pkCol = side.getPrimaryKey().get(0); + Object pkVal = columns.get(pkCol); MapSqlParameterSource params = new MapSqlParameterSource(); - - for (String col : allCols) { - colJoiner.add(col); - valJoiner.add(":" + col); - // 主键不参与 UPDATE 部分 - if (!side.getPrimaryKey().contains(col)) { - updateJoiner.add(col + " = VALUES(" + col + ")"); + allCols.forEach(col -> params.addValue(col, columns.get(col))); + params.addValue("__pk", pkVal); + + // 严禁使用 ON DUPLICATE KEY UPDATE： + // 若表上存在除主键外的唯一索引(如 enterprise_information.unique(enterprise_number)), + // 冲突时 MySQL 会把"另一行"整行改写,造成静默数据覆盖且无法从同步日志回溯。 + // 改为按主键判存在 → UPDATE ... WHERE pk / INSERT 分离,唯一索引冲突时直接抛 DuplicateKeyException, + // 由 TableSyncExecutor 记入 biz_sync_log 并最终置 DEAD,交人工裁决。 + Integer exists = namedJdbcTemplate.queryForObject( + "SELECT COUNT(1) FROM " + side.getTable() + " WHERE " + pkCol + " = :__pk", + new MapSqlParameterSource("__pk", pkVal), Integer.class); + + if (exists != null && exists > 0) { + StringJoiner setJoiner = new StringJoiner(", "); + for (String col : allCols) { + if (!side.getPrimaryKey().contains(col)) { + setJoiner.add(col + " = :" + col); + } + } + if (setJoiner.length() == 0) { + return; + } + String sql = "UPDATE " + side.getTable() + " SET " + setJoiner + " WHERE " + pkCol + " = :__pk"; + log.debug("[BizSync] V1 update SQL: {}", sql); + namedJdbcTemplate.update(sql, params); + } else { + StringJoiner colJoiner = new StringJoiner(", "); + StringJoiner valJoiner = new StringJoiner(", "); + for (String col : allCols) { + colJoiner.add(col); + valJoiner.add(":" + col); + } + String sql = "INSERT INTO " + side.getTable() + " (" + colJoiner + ") VALUES (" + valJoiner + ")"; + log.debug("[BizSync] V1 insert SQL: {}", sql); + try { + namedJdbcTemplate.update(sql, params); + } catch (DuplicateKeyException e) { + handleDuplicateKey(side, pkCol, pkVal, e); } - params.addValue(col, columns.get(col)); } + } - String sql = "INSERT INTO " + side.getTable() + " (" + colJoiner + ") VALUES (" + valJoiner + ")"; - if (updateJoiner.length() > 0) { - sql += " ON DUPLICATE KEY UPDATE " + updateJoiner; + /** + * 区分主键重复（幂等）与唯一键冲突（业务键撞车） + *

主键重复：说明行已存在，视为幂等成功，忽略； + * 唯一键冲突：说明撞到了表上其他唯一索引（如 enterprise_information.unique(enterprise_number)）， + * 抛 UniqueKeyConflictException 交由 Executor 层记入 biz_sync_log 并跳过，不重试。

+ */ + private void handleDuplicateKey(SyncTableSideConfig side, String pkCol, Object pkVal, DuplicateKeyException e) { + Integer recheck = namedJdbcTemplate.queryForObject( + "SELECT COUNT(1) FROM " + side.getTable() + " WHERE " + pkCol + " = :__pk", + new MapSqlParameterSource("__pk", pkVal), Integer.class); + if (recheck != null && recheck > 0) { + log.debug("[BizSync] V1 INSERT 命中幂等重复(主键已存在),忽略: table={}, pk={}", + side.getTable(), pkVal); + return; } - - log.debug("[BizSync] V1 upsert SQL: {}", sql); - namedJdbcTemplate.update(sql, params); + String msg = String.format("唯一键冲突: table=%s, pk=%s, 原因=%s", + side.getTable(), pkVal, e.getMessage()); + log.warn("[BizSync] V1 {}", msg); + throw new UniqueKeyConflictException(side.getTable(), String.valueOf(pkVal), msg, e); } private void doDelete(SyncTableSideConfig side, String pkValue) { diff --git a/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/writer/V2SinkWriter.java b/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/writer/V2SinkWriter.java index fa74f565..0371bb84 100644 --- a/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/writer/V2SinkWriter.java +++ b/zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/writer/V2SinkWriter.java @@ -5,8 +5,10 @@ import com.zdxt.enforcementcode.businessdatasyn.config.SyncLogicalDeleteConfig; import com.zdxt.enforcementcode.businessdatasyn.config.SyncTableConfig; import com.zdxt.enforcementcode.businessdatasyn.config.SyncTableSideConfig; import com.zdxt.enforcementcode.businessdatasyn.domain.dto.RowSnapshot; +import com.zdxt.enforcementcode.businessdatasyn.exception.UniqueKeyConflictException; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.dao.DuplicateKeyException; import org.springframework.jdbc.core.namedparam.MapSqlParameterSource; import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate; import org.springframework.stereotype.Service; @@ -69,29 +71,72 @@ public class V2SinkWriter implements SinkWriter { throw new IllegalStateException("Upsert 没有有效列: " + side.getTable()); } - StringJoiner colJoiner = new StringJoiner(", "); - StringJoiner valJoiner = new StringJoiner(", "); - StringJoiner updateJoiner = new StringJoiner(", "); + String pkCol = side.getPrimaryKey().get(0); + Object pkVal = columns.get(pkCol); MapSqlParameterSource params = new MapSqlParameterSource(); - - for (String col : allCols) { - colJoiner.add(col); - valJoiner.add(":" + col); - // 主键和 v2 独有默认值列不进入 UPDATE 子句：默认值只用于首次插入,更新时保留 V2 已有值 - if (!side.getPrimaryKey().contains(col) && !insertOnlyCols.contains(col)) { - updateJoiner.add(col + " = VALUES(" + col + ")"); + allCols.forEach(col -> params.addValue(col, columns.get(col))); + params.addValue("__pk", pkVal); + + // 严禁使用 ON DUPLICATE KEY UPDATE： + // 若表上存在除主键外的唯一索引(如 enterprise_information.unique(enterprise_number)), + // 冲突时 MySQL 会把"另一行"整行改写,造成静默数据覆盖且无法从同步日志回溯。 + // 改为按主键判存在 → UPDATE ... WHERE pk / INSERT 分离,唯一索引冲突时直接抛 DuplicateKeyException, + // 由 TableSyncExecutor 记入 biz_sync_log 并最终置 DEAD,交人工裁决。 + Integer exists = namedJdbcTemplate.queryForObject( + "SELECT COUNT(1) FROM " + side.getTable() + " WHERE " + pkCol + " = :__pk", + new MapSqlParameterSource("__pk", pkVal), Integer.class); + + if (exists != null && exists > 0) { + StringJoiner setJoiner = new StringJoiner(", "); + for (String col : allCols) { + // 主键不更新；v2 独有默认值列只在 INSERT 生效,UPDATE 时保留 V2 已有值 + if (!side.getPrimaryKey().contains(col) && !insertOnlyCols.contains(col)) { + setJoiner.add(col + " = :" + col); + } + } + if (setJoiner.length() == 0) { + return; + } + String sql = "UPDATE " + side.getTable() + " SET " + setJoiner + " WHERE " + pkCol + " = :__pk"; + log.debug("[BizSync] V2 update SQL: {}", sql); + namedJdbcTemplate.update(sql, params); + } else { + StringJoiner colJoiner = new StringJoiner(", "); + StringJoiner valJoiner = new StringJoiner(", "); + for (String col : allCols) { + colJoiner.add(col); + valJoiner.add(":" + col); + } + String sql = "INSERT INTO " + side.getTable() + " (" + colJoiner + ") VALUES (" + valJoiner + ")"; + log.debug("[BizSync] V2 insert SQL: {}", sql); + try { + namedJdbcTemplate.update(sql, params); + } catch (DuplicateKeyException e) { + handleDuplicateKey(side, pkCol, pkVal, e); } - params.addValue(col, columns.get(col)); } + } - String sql = "INSERT INTO " + side.getTable() + " (" + colJoiner + ") VALUES (" + valJoiner + ")"; - if (updateJoiner.length() > 0) { - sql += " ON DUPLICATE KEY UPDATE " + updateJoiner; + /** + * 区分主键重复（幂等）与唯一键冲突（业务键撞车） + *

主键重复：说明行已存在，视为幂等成功，忽略； + * 唯一键冲突：说明撞到了表上其他唯一索引（如 enterprise_information.unique(enterprise_number)）， + * 抛 UniqueKeyConflictException 交由 Executor 层记入 biz_sync_log 并跳过，不重试。

+ */ + private void handleDuplicateKey(SyncTableSideConfig side, String pkCol, Object pkVal, DuplicateKeyException e) { + Integer recheck = namedJdbcTemplate.queryForObject( + "SELECT COUNT(1) FROM " + side.getTable() + " WHERE " + pkCol + " = :__pk", + new MapSqlParameterSource("__pk", pkVal), Integer.class); + if (recheck != null && recheck > 0) { + log.debug("[BizSync] V2 INSERT 命中幂等重复(主键已存在),忽略: table={}, pk={}", + side.getTable(), pkVal); + return; } - - log.debug("[BizSync] V2 upsert SQL: {}", sql); - namedJdbcTemplate.update(sql, params); + String msg = String.format("唯一键冲突: table=%s, pk=%s, 原因=%s", + side.getTable(), pkVal, e.getMessage()); + log.warn("[BizSync] V2 {}", msg); + throw new UniqueKeyConflictException(side.getTable(), String.valueOf(pkVal), msg, e); } private void doDelete(SyncTableSideConfig side, String pkValue) {