增量同步 – 晚风初歇

增量同步是一种只同步“发生变化的数据”（新增和更新），而不搬运全部数据的机制。相比每次全量覆盖，它能大幅减少数据传输量，适合数据量大或对实时性有要求的场景。

核心原理与局限

它只抓取自上次同步以来产生变动的数据。不过，单纯的增量同步通常无法捕获删除操作（因为被删的数据在源端已不存在），删除往往需要依赖全量同步或特殊的日志解析来补充。

常见实现方式

根据源系统的能力和架构，主要有三种主流做法：

1. 基于时间戳或增量字段（最简单）
在源表中找一个能标识更新时间的字段（如 update_time、gmt_modify）。同步任务每次只查询该字段大于“上次同步时间点”的数据。

优点：实现简单，无需改源库结构。
缺点：无法捕获物理删除；若源表没有更新时间字段则无法使用；高并发下可能存在微小的时间误差导致漏数。

2. 基于触发器（对源库有侵入）
在源库建立数据库触发器，当发生增删改时，自动将变更写入一张专门的日志表，同步工具再去读日志表。

优点：能捕获删除和更新。
缺点：对源库性能有影响，且需要修改源库 schema，很多业务场景不允许这样做。

3. 基于日志解析（CDC，最可靠）
这是目前生产环境主流的高阶方案。通过读取数据库自身的日志（如 MySQL 的 binlog、Oracle 的 redo log），解析出所有的增删改操作并应用到目标端。

优点：对源库几乎无侵入，能完整捕获增删改，且支持断点续传（记录位点，重启后接着读）。
缺点：架构复杂，通常需要借助 Debezium、Canal 等专业工具或云服务。

关键实现细节：全量与增量的衔接

在初次搭建时，通常需要先做一次全量同步，然后无缝切换到增量。
严谨的做法是：在全量快照开始的那一刻，记录下日志的位点或时间戳；等全量搬完后，从那个记录点开始补增量数据，避免中间出现数据断层。

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
         http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.example</groupId>
    <artifactId>incremental-sync</artifactId>
    <version>1.0.0</version>

    <properties>
        <maven.compiler.source>11</maven.compiler.source>
        <maven.compiler.target>11</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

    <dependencies>
        <!-- SQLite JDBC -->
        <dependency>
            <groupId>org.xerial</groupId>
            <artifactId>sqlite-jdbc</artifactId>
            <version>3.45.3.0</version>
        </dependency>
        <!-- Jackson 用于位点 JSON 读写 -->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.17.1</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <!-- 打成可执行 jar，方便直接 java -jar 运行 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.5.1</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals><goal>shade</goal></goals>
                        <configuration>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>com.example.sync.IncrementalSyncDemo</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

`IncrementalSyncDemo.java`

package com.example.sync;

import com.fasterxml.jackson.databind.ObjectMapper;

import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.sql.*;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class IncrementalSyncDemo {
private static final String SOURCE_DB = "source.db";
private static final String TARGET_DB = "target.db";
private static final Path CHECKPOINT_FILE = Paths.get("checkpoint.json");
private static final String INITIAL_TS = "1970-01-01 00:00:00";

private static final DateTimeFormatter FMT =
        DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
private static final ObjectMapper MAPPER = new ObjectMapper();

// ---------- 初始化 ----------

private static void initSource() throws SQLException {
    try (Connection conn = DriverManager.getConnection("jdbc:sqlite:" + SOURCE_DB);
         Statement st = conn.createStatement()) {
        st.execute("""
            CREATE TABLE IF NOT EXISTS source_users (
                id INTEGER PRIMARY KEY,
                name TEXT,
                update_time TEXT
            )
        """);
    }
}

private static void initTarget() throws SQLException {
    try (Connection conn = DriverManager.getConnection("jdbc:sqlite:" + TARGET_DB);
         Statement st = conn.createStatement()) {
        st.execute("""
            CREATE TABLE IF NOT EXISTS target_users (
                id INTEGER PRIMARY KEY,
                name TEXT,
                update_time TEXT
            )
        """);
    }
}

// ---------- 位点读写 ----------

@SuppressWarnings("unchecked")
private static String loadCheckpoint() {
    try {
        if (Files.exists(CHECKPOINT_FILE)) {
            Map<String, String> data = MAPPER.readValue(CHECKPOINT_FILE.toFile(), Map.class);
            String ts = data.get("last_update_time");
            if (ts != null && !ts.isEmpty()) return ts;
        }
    } catch (Exception e) {
        System.err.println("[checkpoint] 读取失败，使用初始位点: " + e.getMessage());
    }
    return INITIAL_TS;
}

private static void saveCheckpoint(String ts) {
    try {
        Map<String, String> data = new HashMap<>();
        data.put("last_update_time", ts);
        MAPPER.writeValue(CHECKPOINT_FILE.toFile(), data);
    } catch (Exception e) {
        throw new RuntimeException("保存位点失败", e);
    }
}

// ---------- 核心：增量同步 ----------

public static void incrementalSync() throws SQLException {
    String lastTs = loadCheckpoint();
    System.out.println("[sync] 上次同步位点: " + lastTs);

    List<Object[]> rows = new ArrayList<>();

    // 1. 从源库按位点拉增量
    try (Connection src = DriverManager.getConnection("jdbc:sqlite:" + SOURCE_DB);
         PreparedStatement ps = src.prepareStatement(
                 "SELECT id, name, update_time FROM source_users " +
                 "WHERE update_time > ? ORDER BY update_time")) {
        ps.setString(1, lastTs);
        try (ResultSet rs = ps.executeQuery()) {
            while (rs.next()) {
                rows.add(new Object[]{
                        rs.getInt("id"),
                        rs.getString("name"),
                        rs.getString("update_time")
                });
            }
        }
    }

    if (rows.isEmpty()) {
        System.out.println("[sync] 无新增数据");
        return;
    }

    // 2. 写入目标库（UPSERT），同时推进位点
    String maxTs = lastTs;
    try (Connection tgt = DriverManager.getConnection("jdbc:sqlite:" + TARGET_DB);
         PreparedStatement ps = tgt.prepareStatement(
                 "INSERT INTO target_users (id, name, update_time) VALUES (?, ?, ?) " +
                 "ON CONFLICT(id) DO UPDATE SET " +
                 "name = excluded.name, update_time = excluded.update_time")) {

        for (Object[] row : rows) {
            ps.setInt(1, (Integer) row[0]);
            ps.setString(2, (String) row[1]);
            ps.setString(3, (String) row[2]);
            ps.addBatch();

            String ts = (String) row[2];
            if (ts.compareTo(maxTs) > 0) maxTs = ts;
        }
        ps.executeBatch();
    }

    saveCheckpoint(maxTs);
    System.out.println("[sync] 同步 " + rows.size() + " 条，新位点: " + maxTs);
}

// ---------- 模拟源库变化 ----------

private static void simulateSourceChanges() throws SQLException {
    String now = LocalDateTime.now().format(FMT);
    try (Connection conn = DriverManager.getConnection("jdbc:sqlite:" + SOURCE_DB);
         PreparedStatement ps = conn.prepareStatement(
                 "INSERT OR REPLACE INTO source_users (id, name, update_time) VALUES (?, ?, ?)")) {
        ps.setInt(1, 1);
        ps.setString(2, "Alice");
        ps.setString(3, now);
        ps.addBatch();

        ps.setInt(1, 2);
        ps.setString(2, "Bob");
        ps.setString(3, now);
        ps.addBatch();

        ps.executeBatch();
    }
    System.out.println("[source] 写入/更新 2 条，时间: " + now);
}

private static void updateOneInSource() throws SQLException {
    String now = LocalDateTime.now().format(FMT);
    try (Connection conn = DriverManager.getConnection("jdbc:sqlite:" + SOURCE_DB);
         PreparedStatement ps = conn.prepareStatement(
                 "UPDATE source_users SET name = ?, update_time = ? WHERE id = 1")) {
        ps.setString(1, "Alice-Updated");
        ps.setString(2, now);
        ps.executeUpdate();
    }
    System.out.println("[source] 更新 id=1，时间: " + now);
}

private static void printTarget() throws SQLException {
    try (Connection conn = DriverManager.getConnection("jdbc:sqlite:" + TARGET_DB);
         Statement st = conn.createStatement();
         ResultSet rs = st.executeQuery("SELECT id, name, update_time FROM target_users ORDER BY id")) {
        while (rs.next()) {
            System.out.printf("  id=%d, name=%s, update_time=%s%n",
                    rs.getInt("id"), rs.getString("name"), rs.getString("update_time"));
        }
    }
}

// ---------- main ----------

public static void main(String[] args) throws Exception {
    initSource();
    initTarget();

    System.out.println("=== 第一次：模拟源库有数据后同步 ===");
    simulateSourceChanges();
    incrementalSync();

    System.out.println("\n=== 第二次：源库无变化，再同步 ===");
    incrementalSync();

    System.out.println("\n=== 第三次：源库更新一条后再同步 ===");
    // 保证 update_time 比上次大（SQLite 时间字符串按字典序比较）
    Thread.sleep(1100);
    updateOneInSource();
    incrementalSync();

    System.out.println("\n=== 目标库最终数据 ===");
    printTarget();
}
}

核心原理与局限

常见实现方式

关键实现细节：全量与增量的衔接

pom.xml

IncrementalSyncDemo.java

发送评论 编辑评论

推荐文章

`IncrementalSyncDemo.java`

发送评论编辑评论