zhuibobo
/
EnforcementCode-2.0
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

更新主键冲突策略

dev-V1.10.2-temp
zhuibobo 3 months ago
parent
06ffe8914d
commit
1ea226ec6f
5 changed files with 169 additions and 37 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 4
      zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/domain/dto/SyncActionEnum.java
  2. 34
      zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/exception/UniqueKeyConflictException.java
  3. 9
      zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/executor/TableSyncExecutor.java
  4. 80
      zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/writer/V1SinkWriter.java
  5. 79
      zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/writer/V2SinkWriter.java

4
zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/domain/dto/SyncActionEnum.java
View File

@ -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;

34
zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/exception/UniqueKeyConflictException.java
View File

@ -0,0 +1,34 @@
package com.zdxt.enforcementcode.businessdatasyn.exception;
/**
* 唯一键冲突异常（非主键冲突）
* <p>用于标识 INSERT 时撞到了表上其他唯一索引（如 enterprise_information.unique(enterprise_number)），
* 而非主键重复。Executor 层识别后记入 biz_sync_log 并跳过，不增加 retry_count。</p>
*
* @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;
}
}

9
zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/executor/TableSyncExecutor.java
View File

@ -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);

80
zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/writer/V1SinkWriter.java
View File

@ -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;
/**
* 区分主键重复（幂等）与唯一键冲突（业务键撞车）
* <p>主键重复：说明行已存在，视为幂等成功，忽略；
* 唯一键冲突：说明撞到了表上其他唯一索引（如 enterprise_information.unique(enterprise_number)），
* 抛 UniqueKeyConflictException 交由 Executor 层记入 biz_sync_log 并跳过，不重试。</p>
*/
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) {

79
zdxt-modules/zdxt-enforcement-code/src/main/java/com/zdxt/enforcementcode/businessdatasyn/writer/V2SinkWriter.java
View File

@ -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;
/**
* 区分主键重复（幂等）与唯一键冲突（业务键撞车）
* <p>主键重复：说明行已存在，视为幂等成功，忽略；
* 唯一键冲突：说明撞到了表上其他唯一索引（如 enterprise_information.unique(enterprise_number)），
* 抛 UniqueKeyConflictException 交由 Executor 层记入 biz_sync_log 并跳过，不重试。</p>
*/
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) {

Write Preview
Loading…
Cancel
Save