manager/c-api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

新增MySQL和pg数据库的同步

Browse Source
This commit is contained in:
gaoxq
2025-08-27 16:39:54 +08:00
parent ff5006a660
commit 306c93e9ad
8 changed files with 352 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

94
src/main/java/com/mini/capi/job/taskDbSync.java
View File

@@ -2,7 +2,9 @@ package com.mini.capi.job;
import com.mini.capi.biz.domain.DbConfig;
import com.mini.capi.biz.domain.SyncTask;
import com.mini.capi.biz.domain.SyncTaskLog;
import com.mini.capi.biz.service.DbConfigService;
import com.mini.capi.biz.service.SyncTaskLogService;
import com.mini.capi.biz.service.SyncTaskService;
import com.mini.capi.config.DataSourceConfig;
import com.mini.capi.model.ApiResult;
@@ -15,9 +17,12 @@ import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.sql.*;
import java.time.Duration;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;
@@ -32,6 +37,9 @@ public class taskDbSync {
@Resource
private DbConfigService dbConfigService;
@Resource
private SyncTaskLogService taskLogService;
@Resource
private Executor hostExecutor;
@@ -80,7 +88,7 @@ public class taskDbSync {
public void syncTableData(SyncTask task, JdbcTemplate sourceJdbc, JdbcTemplate targetJdbc) {
try {
// 确保源表和目标表名转为小写
String sourceTable = task.getSourceTable().toLowerCase();
String sourceTable = task.getSourceTable();
String targetTable = task.getTargetTable().toLowerCase();
// 1. 检查并创建目标表
if (!tableExists(targetJdbc, targetTable)) {
@@ -89,8 +97,7 @@ public class taskDbSync {
// 2. 清空目标表当前ds值的数据
clearTargetTableData(targetJdbc, targetTable);
// 3. 全量同步数据
syncAllData(sourceJdbc, targetJdbc, sourceTable, targetTable);
System.out.println("表 " + sourceTable + " 同步到 " + targetTable + " 成功");
syncAllData(task, sourceJdbc, targetJdbc, sourceTable, targetTable);
} catch (Exception e) {
System.err.println("表同步失败: " + e.getMessage());
}
@@ -255,37 +262,60 @@ public class taskDbSync {
/**
* 全量同步数据
*/
private void syncAllData(JdbcTemplate sourceJdbc, JdbcTemplate targetJdbc,
private void syncAllData(SyncTask task, JdbcTemplate sourceJdbc, JdbcTemplate targetJdbc,
String sourceTable, String targetTable) {
// 1. 获取源表所有数据
String selectSql = "SELECT * FROM " + sourceTable;
List<Map<String, Object>> dataList = sourceJdbc.queryForList(selectSql);
if (dataList.isEmpty()) {
System.out.println("源表 " + sourceTable + " 没有数据需要同步");
return;
}
// 2. 构建插入SQL
List<String> columns = getTableColumnsWithoutException(sourceJdbc, sourceTable);
String insertSql = buildInsertSql(targetTable, columns);
// 3. 批量插入数据
targetJdbc.batchUpdate(insertSql, new BatchPreparedStatementSetter() {
@Override
public void setValues(PreparedStatement ps, int i) throws SQLException {
Map<String, Object> row = dataList.get(i);
int paramIndex = 1;
// 设置原有字段值
for (String column : columns) {
ps.setObject(paramIndex++, row.get(column));
LocalDateTime startTime = LocalDateTime.now();
int totalRows = 0;
int successRows = 0;
int failRows = 0;
String ustatus = "1";
String errorMsg = null;
try {
// 1. 获取源表所有数据
String selectSql = "SELECT * FROM " + sourceTable;
List<Map<String, Object>> dataList = sourceJdbc.queryForList(selectSql);
if (dataList.isEmpty()) {
System.out.println("源表 " + sourceTable + " 没有数据需要同步");
return;
}
// 2. 构建插入SQL
List<String> columns = getTableColumnsWithoutException(sourceJdbc, sourceTable);
String insertSql = buildInsertSql(targetTable, columns);
// 3. 批量插入数据
int[] batchResult = targetJdbc.batchUpdate(insertSql, new BatchPreparedStatementSetter() {
@Override
public void setValues(PreparedStatement ps, int i) throws SQLException {
Map<String, Object> row = dataList.get(i);
int paramIndex = 1;
// 设置原有字段值
for (String column : columns) {
ps.setObject(paramIndex++, row.get(column));
}
// 设置ds字段值
ps.setString(paramIndex, dsValue);
}
// 设置ds字段值
ps.setString(paramIndex, dsValue);
}
@Override
public int getBatchSize() {
return dataList.size();
}
});
System.out.println("成功同步 " + dataList.size() + " 条数据到表 " + targetTable);
@Override
public int getBatchSize() {
return dataList.size();
}
});
// 统计成功/失败行数
successRows = (int) Arrays.stream(batchResult).filter(row -> row >= 0).count();
failRows = totalRows - successRows;
} catch (Exception e) {
ustatus = "2";
errorMsg = e.getMessage() != null ? e.getMessage() : e.getClass().getSimpleName();
System.err.println("同步数据失败: " + errorMsg);
}
DbConfig sourceDbConfig = dbConfigService.getById(task.getSourceDbId());
DbConfig targetDbConfig = dbConfigService.getById(task.getTargetDbId());
LocalDateTime endTime = LocalDateTime.now();
SyncTaskLog taskLog = new SyncTaskLog(task.getTaskId(), task.getTaskName(), task.getSourceDbId(),
sourceDbConfig.getDbName(), sourceTable, task.getTargetDbId(), targetDbConfig.getDbName(), targetTable,
LocalDateTime.now(), LocalDateTime.now(), (long) totalRows, (long) successRows, (long) failRows, ustatus,
errorMsg, (int) Duration.between(startTime, endTime).getSeconds(), "0");
taskLogService.save(taskLog);
}
/**