From be328e3107a6f840637f78bad31d942d42d33e78 Mon Sep 17 00:00:00 2001 From: tangwei Date: Thu, 13 Aug 2026 16:13:43 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BC=98=E5=8C=96=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../yfd/platform/config/SecurityConfig.java | 1 + .../datasource/DynamicDataSource.java | 8 +- .../fp/service/impl/FpRunServiceImpl.java | 4 +- .../system/controller/DataSyncController.java | 93 +++ .../system/domain/DataSyncRequest.java | 49 ++ .../com/yfd/platform/utils/DataSyncUtil.java | 637 ++++++++++++++++++ 6 files changed, 789 insertions(+), 3 deletions(-) create mode 100644 backend/src/main/java/com/yfd/platform/system/controller/DataSyncController.java create mode 100644 backend/src/main/java/com/yfd/platform/system/domain/DataSyncRequest.java create mode 100644 backend/src/main/java/com/yfd/platform/utils/DataSyncUtil.java diff --git a/backend/src/main/java/com/yfd/platform/config/SecurityConfig.java b/backend/src/main/java/com/yfd/platform/config/SecurityConfig.java index 38674041..04ab7190 100644 --- a/backend/src/main/java/com/yfd/platform/config/SecurityConfig.java +++ b/backend/src/main/java/com/yfd/platform/config/SecurityConfig.java @@ -71,6 +71,7 @@ public class SecurityConfig { // .requestMatchers("/system/**").permitAll() // .requestMatchers("/qgcExport/**").permitAll() .requestMatchers("/eng/**").permitAll() + .requestMatchers("/system/**").permitAll() // .requestMatchers("/eq/**").permitAll() // .requestMatchers("/env/**").permitAll() // .requestMatchers("/warn/**").permitAll() diff --git a/backend/src/main/java/com/yfd/platform/datasource/DynamicDataSource.java b/backend/src/main/java/com/yfd/platform/datasource/DynamicDataSource.java index 8b525217..79dc807c 100644 --- a/backend/src/main/java/com/yfd/platform/datasource/DynamicDataSource.java +++ b/backend/src/main/java/com/yfd/platform/datasource/DynamicDataSource.java @@ -36,5 +36,11 @@ public class DynamicDataSource extends AbstractRoutingDataSource { contextHolder.remove(); } - + /** + * 根据数据源 key 获取目标数据源实例（如 dm-master / oracle-master） + */ + public DataSource getDataSourceByKey(String key) { + Map resolved = getResolvedDataSources(); + return resolved == null ? null : resolved.get(key); + } } diff --git a/backend/src/main/java/com/yfd/platform/qgc_env/fp/service/impl/FpRunServiceImpl.java b/backend/src/main/java/com/yfd/platform/qgc_env/fp/service/impl/FpRunServiceImpl.java index ac2c73dc..aba8ad7f 100644 --- a/backend/src/main/java/com/yfd/platform/qgc_env/fp/service/impl/FpRunServiceImpl.java +++ b/backend/src/main/java/com/yfd/platform/qgc_env/fp/service/impl/FpRunServiceImpl.java @@ -2592,7 +2592,7 @@ public class FpRunServiceImpl implements FpRunService { " t2.pycnt AS pycnt, " + " NULL AS pyRate, " + " t.STTP AS sttpCode, " + - " t1.planFtp AS planFtp, " + + " t.ZYGYDXMS AS planFtp, " + " t.ZYGYDX AS fallbackFishJson, " + " NVL(t.ZYGYDXMS, t.JGGYDXMS) AS fallbackFishText " + "FROM SD_FPSS_B_H t " + @@ -3502,7 +3502,7 @@ public class FpRunServiceImpl implements FpRunService { private String buildFpssrlQdrtpSOrderBySql(List sortList) { if (CollUtil.isEmpty(sortList)) { - return " ORDER BY t.YEAR DESC, t.MONTH DESC, t.DR DESC"; + return " ORDER BY t.YEAR DESC, t.MONTH DESC"; } List orders = new ArrayList<>(); for (DataSourceRequest.SortDescriptor sort : sortList) { diff --git a/backend/src/main/java/com/yfd/platform/system/controller/DataSyncController.java b/backend/src/main/java/com/yfd/platform/system/controller/DataSyncController.java new file mode 100644 index 00000000..fb35cc5c --- /dev/null +++ b/backend/src/main/java/com/yfd/platform/system/controller/DataSyncController.java @@ -0,0 +1,93 @@ +package com.yfd.platform.system.controller; + +import cn.hutool.core.util.StrUtil; +import com.yfd.platform.config.ResponseResult; +import com.yfd.platform.system.domain.DataSyncRequest; +import com.yfd.platform.utils.DataSyncUtil; +import io.swagger.v3.oas.annotations.Operation; +import io.swagger.v3.oas.annotations.tags.Tag; +import jakarta.annotation.Resource; +import org.springframework.validation.annotation.Validated; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +/** + * 数据同步控制器：支持 DM / Oracle 之间数据同步（MERGE INTO 方式） + */ +@RestController +@RequestMapping("/system/dataSync") +@Tag(name = "数据同步") +@Validated +public class DataSyncController { + + @Resource + private DataSyncUtil dataSyncUtil; + + /** + * DM 数据同步到 Oracle（使用调用方提供的连接信息） + *

+ * 请求示例（把 DM 库 SD_FPSS_R 同步到 Oracle 库 SD_FPSS_R）： + * 

+     * {
+     *   "sourceTable": "SD_FPSS_R",
+     *   "targetTable": "SD_FPSS_R",
+     *   "keyColumns": ["ID"],
+     *   "clearTarget": true,
+     *   "batchSize": 500,
+     *   "sourceConn": {
+     *     "url": "jdbc:dm://172.16.21.143:5236/QGC_REFA_TEST",
+     *     "username": "QGC_REFA_TEST",
+     *     "password": "xxxx"
+     *   },
+     *   "targetConn": {
+     *     "url": "jdbc:oracle:thin:@172.16.31.172:1521/SDLYZ",
+     *     "username": "SDLY_QX",
+     *     "password": "xxxx"
+     *   }
+     * }
+     *
+ */ + @PostMapping("/dmToOracle") + @Operation(summary = "DM 数据同步到 Oracle（自定义连接，MERGE INTO）") + public ResponseResult dmToOracle(@RequestBody DataSyncRequest request) { + DataSyncUtil.DataSyncResult result = dataSyncUtil.syncWithConnections(buildConfig(request), + request.getSourceConn(), request.getTargetConn()); + return ResponseResult.successData(result); + } + + /** + * 通用数据同步（自定义连接，可指定任意方向） + */ + @PostMapping("/sync") + @Operation(summary = "通用数据同步（自定义连接，MERGE INTO）") + public ResponseResult sync(@RequestBody DataSyncRequest request) { + DataSyncUtil.DataSyncResult result = dataSyncUtil.syncWithConnections(buildConfig(request), + request.getSourceConn(), request.getTargetConn()); + return ResponseResult.successData(result); + } + + private DataSyncUtil.DataSyncConfig buildConfig(DataSyncRequest request) { + if (request == null) { + throw new IllegalArgumentException("请求参数不能为空"); + } + if (StrUtil.isBlank(request.getSourceTable())) { + throw new IllegalArgumentException("源表(sourceTable)不能为空"); + } + if (StrUtil.isBlank(request.getTargetTable())) { + throw new IllegalArgumentException("目标表(targetTable)不能为空"); + } + DataSyncUtil.DataSyncConfig config = new DataSyncUtil.DataSyncConfig(); + config.setSourceTable(request.getSourceTable()); + config.setTargetTable(request.getTargetTable()); + config.setColumns(request.getColumns()); + config.setColumnMapping(request.getColumnMapping()); + config.setKeyColumns(request.getKeyColumns()); + config.setConditionSql(request.getConditionSql()); + config.setOrderColumn(request.getOrderColumn()); + config.setBatchSize(request.getBatchSize()); + config.setClearTarget(request.getClearTarget()); + return config; + } +} diff --git a/backend/src/main/java/com/yfd/platform/system/domain/DataSyncRequest.java b/backend/src/main/java/com/yfd/platform/system/domain/DataSyncRequest.java new file mode 100644 index 00000000..eebf4765 --- /dev/null +++ b/backend/src/main/java/com/yfd/platform/system/domain/DataSyncRequest.java @@ -0,0 +1,49 @@ +package com.yfd.platform.system.domain; + +import com.yfd.platform.utils.DataSyncUtil; +import io.swagger.v3.oas.annotations.media.Schema; +import lombok.Data; + +import java.util.List; +import java.util.Map; + +/** + * 数据同步请求 + */ +@Data +@Schema(description = "数据同步请求") +public class DataSyncRequest { + + @Schema(description = "源表名（DM/Oracle）") + private String sourceTable; + + @Schema(description = "目标表名（可与源表名不同）") + private String targetTable; + + @Schema(description = "源表同步列（不指定则自动读取源表全列）") + private List columns; + + @Schema(description = "字段映射：源列名 -> 目标列名") + private Map columnMapping; + + @Schema(description = "MERGE 匹配主键列（目标表列名，必填）") + private List keyColumns; + + @Schema(description = "增量同步条件 SQL 片段（基于源表列）") + private String conditionSql; + + @Schema(description = "分页排序列（源表列，不指定则取第一列）") + private String orderColumn; + + @Schema(description = "每批同步条数，默认 500") + private Integer batchSize; + + @Schema(description = "同步前是否清空目标表（全量覆盖）") + private Boolean clearTarget; + + @Schema(description = "源库连接信息（driverClassName 不填时按 URL 自动识别）") + private DataSyncUtil.DbConnectionInfo sourceConn; + + @Schema(description = "目标库连接信息（driverClassName 不填时按 URL 自动识别）") + private DataSyncUtil.DbConnectionInfo targetConn; +} diff --git a/backend/src/main/java/com/yfd/platform/utils/DataSyncUtil.java b/backend/src/main/java/com/yfd/platform/utils/DataSyncUtil.java new file mode 100644 index 00000000..b7417640 --- /dev/null +++ b/backend/src/main/java/com/yfd/platform/utils/DataSyncUtil.java @@ -0,0 +1,637 @@ +package com.yfd.platform.utils; + +import cn.hutool.core.collection.CollUtil; +import cn.hutool.core.util.StrUtil; +import com.alibaba.druid.pool.DruidDataSource; +import com.yfd.platform.datasource.DataSourceKeys; +import com.yfd.platform.datasource.DynamicDataSource; +import jakarta.annotation.Resource; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Component; + +import javax.sql.DataSource; +import java.sql.ResultSetMetaData; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +/** + * 数据同步控制类：支持 Oracle 与达梦(DM) 之间双向同步，使用 MERGE INTO 方式合并数据。 + *

+ * 支持两种数据源来源： + *

    + *
  • 项目配置的数据源：{@link #syncTable(String, String, DataSyncConfig)}（dm-master / oracle-master 等 key）
    • + *
  • 调用方自定义连接：{@link #syncWithConnections(DataSyncConfig, DbConnectionInfo, DbConnectionInfo)}（URL/账号/密码）
    • + *
+ *

+ *

+ * 支持：源表名与目标表名不同、源字段与目标字段不同（通过 {@link DataSyncConfig#setColumnMapping} 配置映射）。 + *

+ */ +@Component +public class DataSyncUtil { + + private static final Logger log = LoggerFactory.getLogger(DataSyncUtil.class); + + @Resource + private DynamicDataSource dynamicDataSource; + + // ==================== 项目数据源入口 ==================== + + /** + * 达梦(DM) → Oracle 数据同步（使用项目配置的 dm-master / oracle-master 数据源） + */ + public DataSyncResult syncDmToOracle(DataSyncConfig config) { + return syncTable(DataSourceKeys.DM_MASTER, DataSourceKeys.ORACLE_MASTER, config); + } + + /** + * Oracle → 达梦(DM) 数据同步（使用项目配置的 oracle-master / dm-master 数据源） + */ + public DataSyncResult syncOracleToDm(DataSyncConfig config) { + return syncTable(DataSourceKeys.ORACLE_MASTER, DataSourceKeys.DM_MASTER, config); + } + + /** + * 通用数据同步：从源数据源读取，MERGE INTO 写入目标数据源（使用项目配置的数据源 key） + * + * @param sourceDsKey 源数据源 key（如 dm-master / oracle-master） + * @param targetDsKey 目标数据源 key + * @param config 同步配置 + */ + public DataSyncResult syncTable(String sourceDsKey, String targetDsKey, DataSyncConfig config) { + DataSource sourceDataSource = dynamicDataSource.getDataSourceByKey(sourceDsKey); + DataSource targetDataSource = dynamicDataSource.getDataSourceByKey(targetDsKey); + if (sourceDataSource == null) { + throw new IllegalArgumentException("未找到数据源: " + sourceDsKey); + } + if (targetDataSource == null) { + throw new IllegalArgumentException("未找到数据源: " + targetDsKey); + } + return doSync(new JdbcTemplate(sourceDataSource), new JdbcTemplate(targetDataSource), config); + } + + // ==================== 自定义连接入口 ==================== + + /** + * 使用调用方提供的连接信息同步数据（DM → Oracle / Oracle → DM 均适用） + *

+ * 典型场景：把 DM 库 {@code SD_FPSS_R} 同步到 Oracle 库 {@code SD_FPSS_R}， + * 连接信息由调用方在请求中提供。 + *

+ * + * @param config 同步配置 + * @param sourceConn 源库连接信息（driverClassName 不填时按 URL 自动识别） + * @param targetConn 目标库连接信息 + */ + public DataSyncResult syncWithConnections(DataSyncConfig config, DbConnectionInfo sourceConn, DbConnectionInfo targetConn) { + if (sourceConn == null || StrUtil.isBlank(sourceConn.getUrl())) { + throw new IllegalArgumentException("源库连接信息(url)不能为空"); + } + if (targetConn == null || StrUtil.isBlank(targetConn.getUrl())) { + throw new IllegalArgumentException("目标库连接信息(url)不能为空"); + } + DruidDataSource sourceDataSource = buildDataSource(sourceConn); + DruidDataSource targetDataSource = buildDataSource(targetConn); + try { + return doSync(new JdbcTemplate(sourceDataSource), new JdbcTemplate(targetDataSource), config); + } finally { + sourceDataSource.close(); + targetDataSource.close(); + } + } + + // ==================== 核心同步逻辑 ==================== + + /** + * 核心同步：从源 JdbcTemplate 分页读取，MERGE INTO 写入目标 JdbcTemplate + */ + private DataSyncResult doSync(JdbcTemplate sourceJdbc, JdbcTemplate targetJdbc, DataSyncConfig config) { + long startTime = System.currentTimeMillis(); + DataSyncResult result = new DataSyncResult(); + result.setStartTime(startTime); + try { + validateConfig(config); + + // 1. 源表列（未指定时自动读取源表结构） + List sourceColumns = normalizeColumns(config.getColumns()); + if (CollUtil.isEmpty(sourceColumns)) { + sourceColumns = loadTableColumns(sourceJdbc, config.getSourceTable()); + if (CollUtil.isEmpty(sourceColumns)) { + throw new IllegalArgumentException("未从源表 [" + config.getSourceTable() + "] 获取到任何列"); + } + } + + // 2. 目标表列：按 columnMapping 映射，未配置映射的列名保持一致 + Map columnMapping = normalizeMapping(config.getColumnMapping()); + List targetColumns = new ArrayList<>(sourceColumns.size()); + for (String sourceColumn : sourceColumns) { + targetColumns.add(columnMapping.getOrDefault(sourceColumn, sourceColumn)); + } + + // 3. 主键列：keyColumns 配置为目标表列名，反查对应的源表列名用于取值 + List keyColumns = normalizeColumns(config.getKeyColumns()); + Map reverseMapping = new HashMap<>(); + for (Map.Entry entry : columnMapping.entrySet()) { + reverseMapping.put(entry.getValue(), entry.getKey()); + } + List keySourceColumns = new ArrayList<>(keyColumns.size()); + for (String keyColumn : keyColumns) { + keySourceColumns.add(reverseMapping.getOrDefault(keyColumn, keyColumn)); + } + + // 4. 可选：同步前清空目标表（全量覆盖） + if (Boolean.TRUE.equals(config.getClearTarget())) { + clearTargetTable(targetJdbc, config.getTargetTable()); + } + + // 5. 分页读取源数据 → 批量 MERGE 写入目标库 + int batchSize = config.getBatchSize() != null && config.getBatchSize() > 0 + ? config.getBatchSize() : 500; + long totalRead = 0; + long totalProcessed = 0; + int offset = 0; + + while (true) { + List> rows = readPage(sourceJdbc, config, sourceColumns, offset, batchSize); + if (CollUtil.isEmpty(rows)) { + break; + } + int[] effects = mergeInto(targetJdbc, config, sourceColumns, targetColumns, + keyColumns, keySourceColumns, rows); + for (int effect : effects) { + totalProcessed += effect; + } + totalRead += rows.size(); + offset += rows.size(); + if (rows.size() < batchSize) { + break; + } + } + + result.setSuccess(true); + result.setTotalRead(totalRead); + result.setTotalProcessed(totalProcessed); + result.setEndTime(System.currentTimeMillis()); + result.setElapsedMillis(result.getEndTime() - startTime); + result.setMessage("同步完成：读取 " + totalRead + " 条，MERGE 处理 " + totalProcessed + " 条"); + log.info("数据同步完成 sourceTable={} targetTable={} {}", config.getSourceTable(), + config.getTargetTable(), result.getMessage()); + } catch (Exception e) { + result.setSuccess(false); + result.setEndTime(System.currentTimeMillis()); + result.setElapsedMillis(result.getEndTime() - startTime); + result.setMessage("数据同步失败：" + e.getMessage()); + log.error("数据同步失败 sourceTable={} targetTable={}", + config == null ? "" : config.getSourceTable(), + config == null ? "" : config.getTargetTable(), e); + } + return result; + } + + // ==================== 内部实现 ==================== + + private void validateConfig(DataSyncConfig config) { + if (config == null) { + throw new IllegalArgumentException("同步配置(config)不能为空"); + } + if (StrUtil.isBlank(config.getSourceTable())) { + throw new IllegalArgumentException("源表(sourceTable)不能为空"); + } + if (StrUtil.isBlank(config.getTargetTable())) { + throw new IllegalArgumentException("目标表(targetTable)不能为空"); + } + if (CollUtil.isEmpty(config.getKeyColumns())) { + throw new IllegalArgumentException("必须配置主键列(keyColumns)用于 MERGE 匹配"); + } + } + + private List normalizeColumns(List columns) { + if (CollUtil.isEmpty(columns)) { + return new ArrayList<>(); + } + return columns.stream() + .filter(StrUtil::isNotBlank) + .map(String::trim) + .map(String::toUpperCase) + .distinct() + .collect(Collectors.toList()); + } + + private Map normalizeMapping(Map mapping) { + Map normalized = new HashMap<>(); + if (mapping == null || mapping.isEmpty()) { + return normalized; + } + for (Map.Entry entry : mapping.entrySet()) { + if (StrUtil.isNotBlank(entry.getKey()) && StrUtil.isNotBlank(entry.getValue())) { + normalized.put(entry.getKey().trim().toUpperCase(), entry.getValue().trim().toUpperCase()); + } + } + return normalized; + } + + /** + * 根据连接信息构建 Druid 数据源，driverClassName 未填时按 URL 自动识别 + */ + private DruidDataSource buildDataSource(DbConnectionInfo conn) { + DruidDataSource dataSource = new DruidDataSource(); + dataSource.setDriverClassName(resolveDriverClassName(conn)); + dataSource.setUrl(conn.getUrl()); + dataSource.setUsername(conn.getUsername()); + dataSource.setPassword(conn.getPassword()); + dataSource.setInitialSize(1); + dataSource.setMinIdle(1); + dataSource.setMaxActive(5); + dataSource.setMaxWait(30000); + dataSource.setValidationQuery("SELECT 1 FROM DUAL"); + dataSource.setTestWhileIdle(true); + dataSource.setBreakAfterAcquireFailure(true); + dataSource.setConnectionErrorRetryAttempts(0); + try { + dataSource.init(); + } catch (SQLException e) { + dataSource.close(); + throw new IllegalStateException("初始化源数据源失败: " + conn.getUrl(), e); + } + return dataSource; + } + + private String resolveDriverClassName(DbConnectionInfo conn) { + if (StrUtil.isNotBlank(conn.getDriverClassName())) { + return conn.getDriverClassName(); + } + String url = conn.getUrl() == null ? "" : conn.getUrl().toLowerCase(); + if (url.startsWith("jdbc:dm")) { + return "dm.jdbc.driver.DmDriver"; + } + if (url.startsWith("jdbc:oracle")) { + return "oracle.jdbc.OracleDriver"; + } + throw new IllegalArgumentException("无法根据 URL 识别驱动类型，请显式传入 driverClassName: " + conn.getUrl()); + } + + /** + * 读取源表全部列名（大写） + */ + private List loadTableColumns(JdbcTemplate sourceJdbc, String table) { + List columns = new ArrayList<>(); + sourceJdbc.query("SELECT * FROM " + table + " WHERE 1 = 0", rs -> { + ResultSetMetaData metaData = rs.getMetaData(); + for (int i = 1; i <= metaData.getColumnCount(); i++) { + columns.add(metaData.getColumnName(i).toUpperCase()); + } + }); + return columns; + } + + /** + * 分页读取源数据（Oracle/DM 均支持 ROWNUM 分页） + */ + private List> readPage(JdbcTemplate sourceJdbc, DataSyncConfig config, + List sourceColumns, int offset, int batchSize) { + String columnList = String.join(", ", sourceColumns); + String orderBy = StrUtil.isBlank(config.getOrderColumn()) + ? sourceColumns.get(0) : config.getOrderColumn().trim().toUpperCase(); + String where = StrUtil.isBlank(config.getConditionSql()) + ? "" : " AND " + config.getConditionSql(); + String sql = "SELECT * FROM (SELECT t.*, ROWNUM rn FROM (" + + "SELECT " + columnList + " FROM " + config.getSourceTable() + + " WHERE 1 = 1" + where + + " ORDER BY " + orderBy + + ") t WHERE ROWNUM <= " + (offset + batchSize) + + ") WHERE rn > " + offset; + return sourceJdbc.queryForList(sql); + } + + /** + * 批量 MERGE INTO 写入目标库 + * + * @param sourceColumns 源表列（USING 子句占位别名） + * @param targetColumns 目标表列（与 sourceColumns 一一对应） + * @param keyColumns 目标表主键列（ON 匹配 t 侧） + * @param keySourceColumns 源表主键列（ON 匹配 s 侧，与 keyColumns 一一对应） + */ + private int[] mergeInto(JdbcTemplate targetJdbc, DataSyncConfig config, + List sourceColumns, List targetColumns, + List keyColumns, List keySourceColumns, + List> rows) { + String targetTable = config.getTargetTable(); + + // USING 子句：SELECT ? AS SRC_COL1, ? AS SRC_COL2 FROM DUAL + String usingSelect = sourceColumns.stream() + .map(c -> "? AS " + c) + .collect(Collectors.joining(", ")); + + // ON 匹配条件：t.KEY_TARGET = s.KEY_SOURCE + List onItems = new ArrayList<>(keyColumns.size()); + for (int i = 0; i < keyColumns.size(); i++) { + onItems.add("t." + keyColumns.get(i) + " = s." + keySourceColumns.get(i)); + } + String onCondition = String.join(" AND ", onItems); + + // 更新列（排除主键列）：t.TARGET_COL = s.SOURCE_COL + List updateSetItems = new ArrayList<>(); + for (int i = 0; i < sourceColumns.size(); i++) { + String targetColumn = targetColumns.get(i); + if (keyColumns.contains(targetColumn)) { + continue; + } + updateSetItems.add("t." + targetColumn + " = s." + sourceColumns.get(i)); + } + if (updateSetItems.isEmpty()) { + throw new IllegalArgumentException("目标表除主键列外没有可更新的列，无法构造 MERGE 的 WHEN MATCHED 分支"); + } + String updateSet = String.join(", ", updateSetItems); + + // 插入列：INSERT (TARGET_COL1, TARGET_COL2) VALUES (s.SRC_COL1, s.SRC_COL2) + String insertColumns = String.join(", ", targetColumns); + String insertValues = sourceColumns.stream() + .map(c -> "s." + c) + .collect(Collectors.joining(", ")); + + String sql = "MERGE INTO " + targetTable + " t " + + "USING (SELECT " + usingSelect + " FROM DUAL) s " + + "ON (" + onCondition + ") " + + "WHEN MATCHED THEN UPDATE SET " + updateSet + " " + + "WHEN NOT MATCHED THEN INSERT (" + insertColumns + ") VALUES (" + insertValues + ")"; + + List batchArgs = new ArrayList<>(rows.size()); + for (Map row : rows) { + Object[] args = new Object[sourceColumns.size()]; + for (int i = 0; i < sourceColumns.size(); i++) { + args[i] = row.get(sourceColumns.get(i)); + } + batchArgs.add(args); + } + return targetJdbc.batchUpdate(sql, batchArgs); + } + + /** + * 清空目标表（TRUNCATE，DDL 自动提交） + */ + private void clearTargetTable(JdbcTemplate targetJdbc, String targetTable) { + targetJdbc.execute("TRUNCATE TABLE " + targetTable); + log.info("已清空目标表 {}", targetTable); + } + + // ==================== 配置与结果 ==================== + + /** + * 数据库连接信息（调用方自定义连接时使用） + */ + public static class DbConnectionInfo { + + /** JDBC URL，如 jdbc:dm://127.0.0.1:5236/DMDB 或 jdbc:oracle:thin:@127.0.0.1:1521/ORCL */ + private String url; + + /** 用户名 */ + private String username; + + /** 密码 */ + private String password; + + /** 驱动类名，不填时按 URL 自动识别（dm.jdbc.driver.DmDriver / oracle.jdbc.OracleDriver） */ + private String driverClassName; + + public String getUrl() { + return url; + } + + public void setUrl(String url) { + this.url = url; + } + + public String getUsername() { + return username; + } + + public void setUsername(String username) { + this.username = username; + } + + public String getPassword() { + return password; + } + + public void setPassword(String password) { + this.password = password; + } + + public String getDriverClassName() { + return driverClassName; + } + + public void setDriverClassName(String driverClassName) { + this.driverClassName = driverClassName; + } + } + + /** + * 数据同步配置 + */ + public static class DataSyncConfig { + + /** 源表名（DM/Oracle） */ + private String sourceTable; + + /** 目标表名（可与源表名不同） */ + private String targetTable; + + /** 源表同步列（不指定则自动读取源表全列） */ + private List columns; + + /** + * 字段映射：源列名 -> 目标列名。 + * 未在映射中的列默认目标列名与源列名一致； + * 若两边字段名完全相同，可不配置。 + */ + private Map columnMapping; + + /** MERGE 匹配主键列（目标表列名，必填） */ + private List keyColumns; + + /** 增量同步条件 SQL 片段（基于源表列），如 MODIFY_TIME >= TO_DATE('2025-01-01','YYYY-MM-DD') */ + private String conditionSql; + + /** 分页排序列（源表列，不指定则取第一列） */ + private String orderColumn; + + /** 每批同步条数，默认 500 */ + private Integer batchSize; + + /** 同步前是否清空目标表（全量覆盖） */ + private Boolean clearTarget; + + public String getSourceTable() { + return sourceTable; + } + + public void setSourceTable(String sourceTable) { + this.sourceTable = sourceTable; + } + + public String getTargetTable() { + return targetTable; + } + + public void setTargetTable(String targetTable) { + this.targetTable = targetTable; + } + + public List getColumns() { + return columns; + } + + public void setColumns(List columns) { + this.columns = columns; + } + + public Map getColumnMapping() { + return columnMapping; + } + + public void setColumnMapping(Map columnMapping) { + this.columnMapping = columnMapping; + } + + public List getKeyColumns() { + return keyColumns; + } + + public void setKeyColumns(List keyColumns) { + this.keyColumns = keyColumns; + } + + public String getConditionSql() { + return conditionSql; + } + + public void setConditionSql(String conditionSql) { + this.conditionSql = conditionSql; + } + + public String getOrderColumn() { + return orderColumn; + } + + public void setOrderColumn(String orderColumn) { + this.orderColumn = orderColumn; + } + + public Integer getBatchSize() { + return batchSize; + } + + public void setBatchSize(Integer batchSize) { + this.batchSize = batchSize; + } + + public Boolean getClearTarget() { + return clearTarget; + } + + public void setClearTarget(Boolean clearTarget) { + this.clearTarget = clearTarget; + } + } + + /** + * 数据同步结果 + */ + public static class DataSyncResult { + + /** 是否成功 */ + private boolean success; + + /** 读取记录数 */ + private long totalRead; + + /** MERGE 处理记录数（含新增+更新） */ + private long totalProcessed; + + /** 开始时间 */ + private long startTime; + + /** 结束时间 */ + private long endTime; + + /** 耗时（毫秒） */ + private long elapsedMillis; + + /** 结果描述 */ + private String message; + + public boolean isSuccess() { + return success; + } + + public void setSuccess(boolean success) { + this.success = success; + } + + public long getTotalRead() { + return totalRead; + } + + public void setTotalRead(long totalRead) { + this.totalRead = totalRead; + } + + public long getTotalProcessed() { + return totalProcessed; + } + + public void setTotalProcessed(long totalProcessed) { + this.totalProcessed = totalProcessed; + } + + public long getStartTime() { + return startTime; + } + + public void setStartTime(long startTime) { + this.startTime = startTime; + } + + public long getEndTime() { + return endTime; + } + + public void setEndTime(long endTime) { + this.endTime = endTime; + } + + public long getElapsedMillis() { + return elapsedMillis; + } + + public void setElapsedMillis(long elapsedMillis) { + this.elapsedMillis = elapsedMillis; + } + + public String getMessage() { + return message; + } + + public void setMessage(String message) { + this.message = message; + } + + @Override + public String toString() { + return "DataSyncResult{" + + "success=" + success + + ", totalRead=" + totalRead + + ", totalProcessed=" + totalProcessed + + ", elapsedMillis=" + elapsedMillis + + ", message='" + message + '\'' + + '}'; + } + } +}