import java.sql.Connection;
|
import java.sql.DriverManager;
|
import java.sql.PreparedStatement;
|
import java.sql.ResultSet;
|
import java.sql.ResultSetMetaData;
|
import java.sql.Types;
|
import java.util.ArrayList;
|
import java.util.LinkedHashMap;
|
import java.util.List;
|
import java.util.Map;
|
import java.util.TreeMap;
|
|
/**
|
* 两个库按主键做并集：目标库缺的行插入，同主键内容不同的行按「较新者胜」更新。
|
*
|
* <p>用法：java SyncTool <源url> <目标url> [--dry-run]
|
*
|
* <p>只处理两边都存在的业务表；会话、审计、动态码、事件这类瞬时表与仅存在于一侧的遗留表都不动。
|
* 全部使用参数化语句，不拼字符串，避免转义问题。
|
*/
|
public class SyncTool {
|
|
/** 不参与合并的表：瞬时状态、审计流水、固定参考数据与迁移账本。 */
|
private static final List<String> EXCLUDED_SUFFIXES = List.of(
|
"_session", "_audit", "_event", "_log", "_dynamic_code", "_verification_code",
|
"bank_branch_directory", "flyway_schema_history", "merchant_verification_attempt",
|
"merchant_login_audit", "admin_login_audit", "talent_dynamic_code",
|
"order_dynamic_verification_code", "free_stay_verification_code");
|
|
public static void main(String[] args) throws Exception {
|
String sourceUrl = args[0];
|
String targetUrl = args[1];
|
boolean dryRun = false;
|
java.util.Map<String, java.util.Set<String>> rekey = new java.util.HashMap<>();
|
for (String arg : args) {
|
if ("--dry-run".equals(arg)) dryRun = true;
|
// --rekey=表:主键,主键 —— 只对这些「已确认同主键不同实体」的行换新主键插入。
|
if (arg.startsWith("--rekey=")) {
|
String[] parts = arg.substring("--rekey=".length()).split(":");
|
java.util.Set<String> ids = new java.util.HashSet<>();
|
for (String id : parts[1].split(",")) ids.add(id.trim());
|
rekey.put(parts[0].trim(), ids);
|
}
|
}
|
|
try (Connection source = DriverManager.getConnection(sourceUrl, "root", "123456");
|
Connection target = DriverManager.getConnection(targetUrl, "root", "123456")) {
|
List<String> tables = sharedTables(source, target);
|
int totalInserted = 0;
|
int totalConflicts = 0;
|
int totalSkipped = 0;
|
System.out.printf("%-38s %8s %9s %8s%n", "表", "插入", "冲突未动", "插入失败");
|
for (String table : tables) {
|
if (excluded(table)) continue;
|
String key = primaryKey(source, table);
|
if (key == null) continue;
|
List<String> columns = sharedColumns(source, target, table);
|
if (columns == null || columns.isEmpty() || !columns.contains(key)) continue;
|
int[] result = sync(source, target, table, key, columns, dryRun, rekey.get(table));
|
if (result[0] != 0 || result[1] != 0 || result[2] != 0) {
|
System.out.printf("%-38s %8d %9d %8d%n", table, result[0], result[1], result[2]);
|
}
|
totalInserted += result[0];
|
totalConflicts += result[1];
|
totalSkipped += result[2];
|
}
|
System.out.println("合计：插入 " + totalInserted + " 行，冲突未动 " + totalConflicts
|
+ " 行，插入失败 " + totalSkipped + " 行" + (dryRun ? "（干跑，未写入）" : "（已写入）"));
|
}
|
}
|
|
/**
|
* 只做「并集插入」：目标库缺的行按原主键插入；主键存在但内容不同的行一律不覆盖（只计数）。
|
*
|
* <p>冲突行不动是有意的：两个库的同一主键未必是同一实体（各自的登录账号、卡、进件记录会各自增长），
|
* 覆盖任何一侧都会改坏那侧的数据。{@code rekey} 列表里的表是已确认「同主键不同实体」的，
|
* 这类行改用目标库的新主键插入，避免丢掉源侧那条真实存在的数据。
|
*/
|
private static int[] sync(Connection source, Connection target, String table, String key,
|
List<String> columns, boolean dryRun, java.util.Set<String> rekeyIds) throws Exception {
|
Map<String, Object[]> sourceRows = rows(source, table, key, columns);
|
Map<String, Object[]> targetRows = rows(target, table, key, columns);
|
String columnList = String.join(", ", columns.stream().map(c -> "`" + c + "`").toList());
|
String placeholders = String.join(", ", columns.stream().map(c -> "?").toList());
|
int keyIndex = columns.indexOf(key);
|
long nextId = -1;
|
int inserted = 0;
|
int conflicts = 0;
|
int failed = 0;
|
for (Map.Entry<String, Object[]> entry : sourceRows.entrySet()) {
|
Object[] targetRow = targetRows.get(entry.getKey());
|
Object[] values = entry.getValue();
|
String insertId = entry.getKey();
|
if (targetRow != null) {
|
if (sameRow(values, targetRow)) continue;
|
if (rekeyIds == null || !rekeyIds.contains(entry.getKey())) {
|
conflicts++;
|
continue;
|
}
|
// 同主键不同实体：换一个新主键插入，别覆盖目标库那条。只在真要换号时才取 max，
|
// 免得在字符串主键（例如 talent_user_agreement）上做数字解析。
|
if (nextId < 0) nextId = maxId(target, table, key) + 1;
|
insertId = String.valueOf(nextId);
|
values = values.clone();
|
values[keyIndex] = nextId;
|
nextId++;
|
}
|
if (dryRun) {
|
inserted++;
|
continue;
|
}
|
try (PreparedStatement statement = target.prepareStatement(
|
"insert into `" + table + "` (" + columnList + ") values (" + placeholders + ")")) {
|
bind(statement, values, columns);
|
statement.executeUpdate();
|
inserted++;
|
} catch (Exception exception) {
|
failed++;
|
System.out.println(" ! " + table + " 主键 " + insertId + " 插入失败：" + exception.getMessage());
|
}
|
}
|
return new int[]{inserted, conflicts, failed};
|
}
|
|
private static long maxId(Connection connection, String table, String key) throws Exception {
|
try (PreparedStatement statement = connection.prepareStatement(
|
"select coalesce(max(`" + key + "`), 0) from `" + table + "`");
|
ResultSet rows = statement.executeQuery()) {
|
rows.next();
|
return rows.getLong(1);
|
}
|
}
|
|
private static boolean sameRow(Object[] source, Object[] target) {
|
if (source.length != target.length) return false;
|
for (int index = 0; index < source.length; index++) {
|
String left = source[index] == null ? "\u0000" : String.valueOf(source[index]);
|
String right = target[index] == null ? "\u0000" : String.valueOf(target[index]);
|
if (!left.equals(right)) return false;
|
}
|
return true;
|
}
|
|
private static Map<String, Object[]> rows(Connection connection, String table, String key,
|
List<String> columns) throws Exception {
|
Map<String, Object[]> rows = new TreeMap<>();
|
String selection = String.join(", ", columns.stream().map(c -> "`" + c + "`").toList());
|
try (PreparedStatement statement = connection.prepareStatement(
|
"select " + selection + " from `" + table + "`");
|
ResultSet result = statement.executeQuery()) {
|
while (result.next()) {
|
Object[] values = new Object[columns.size()];
|
for (int index = 0; index < columns.size(); index++) values[index] = result.getObject(index + 1);
|
rows.put(String.valueOf(result.getObject(key)), values);
|
}
|
}
|
return rows;
|
}
|
|
private static void bind(PreparedStatement statement, Object[] values, List<String> columns) throws Exception {
|
for (int index = 0; index < values.length; index++) {
|
if (values[index] == null) statement.setNull(index + 1, Types.NULL);
|
else statement.setObject(index + 1, values[index]);
|
}
|
}
|
|
private static boolean excluded(String table) {
|
for (String suffix : EXCLUDED_SUFFIXES) if (table.equals(suffix) || table.endsWith(suffix)) return true;
|
return false;
|
}
|
|
private static List<String> sharedTables(Connection source, Connection target) throws Exception {
|
List<String> sourceTables = tables(source);
|
List<String> targetTables = tables(target);
|
sourceTables.retainAll(targetTables);
|
return sourceTables;
|
}
|
|
private static List<String> tables(Connection connection) throws Exception {
|
List<String> tables = new ArrayList<>();
|
try (PreparedStatement statement = connection.prepareStatement("""
|
select table_name from information_schema.tables
|
where table_schema = database() and table_type = 'BASE TABLE' order by table_name
|
""");
|
ResultSet rows = statement.executeQuery()) {
|
while (rows.next()) tables.add(rows.getString(1));
|
}
|
return tables;
|
}
|
|
private static List<String> sharedColumns(Connection source, Connection target, String table) throws Exception {
|
List<String> columns = columns(source, table);
|
columns.retainAll(columns(target, table));
|
return columns;
|
}
|
|
/**
|
* 可写列：跳过生成列（{@code generated always as ...}）。生成列由数据库按表达式算出来，
|
* 显式写入会被 MySQL 拒绝（V132/V134/V135 都加了生成列）。
|
*/
|
private static List<String> columns(Connection connection, String table) throws Exception {
|
List<String> columns = new ArrayList<>();
|
try (PreparedStatement statement = connection.prepareStatement("""
|
select column_name, extra, generation_expression from information_schema.columns
|
where table_schema = database() and table_name = ? order by ordinal_position
|
""")) {
|
statement.setString(1, table);
|
try (ResultSet rows = statement.executeQuery()) {
|
while (rows.next()) {
|
String extra = rows.getString(2) == null ? "" : rows.getString(2).toUpperCase(java.util.Locale.ROOT);
|
String expression = rows.getString(3);
|
boolean generated = extra.contains("GENERATED")
|
|| (expression != null && !expression.isBlank());
|
if (!generated) columns.add(rows.getString(1));
|
}
|
}
|
}
|
return columns;
|
}
|
|
private static String primaryKey(Connection connection, String table) throws Exception {
|
List<String> keys = new ArrayList<>();
|
try (PreparedStatement statement = connection.prepareStatement("""
|
select column_name from information_schema.key_column_usage
|
where table_schema = database() and table_name = ? and constraint_name = 'PRIMARY'
|
order by ordinal_position
|
""")) {
|
statement.setString(1, table);
|
try (ResultSet rows = statement.executeQuery()) {
|
while (rows.next()) keys.add(rows.getString(1));
|
}
|
}
|
return keys.size() == 1 ? keys.get(0) : null;
|
}
|
}
|
