Java增量同步案例怎么开发

本文目录导读：

1. 基于时间戳的增量同步
2. 基于版本号的增量同步
3. 基于CDC的增量同步（使用Debezium）
4. 完整的增量同步框架
5. 使用示例

我来为你详细介绍Java增量同步的开发案例,包括几种常见的实现方式。

基于时间戳的增量同步

实现方案

/**
 * 数据变更记录实体
 */
@Data
public class ChangeRecord {
    private Long id;
    private String tableName;
    private String recordId;
    private String operationType; // INSERT, UPDATE, DELETE
    private String changeData;    // JSON格式的变更数据
    private Date createTime;
    private Integer syncStatus;   // 0-未同步, 1-已同步
    private Integer version;
}
/**
 * 增量同步服务
 */
@Service
public class IncrementalSyncService {
    @Autowired
    private JdbcTemplate jdbcTemplate;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    /**
     * 获取待同步数据
     */
    public List<ChangeRecord> getPendingSyncRecords(Date lastSyncTime, int batchSize) {
        String sql = "SELECT * FROM change_records WHERE create_time > ? AND sync_status = 0 " +
                    "ORDER BY create_time ASC LIMIT ?";
        return jdbcTemplate.query(sql, new Object[]{lastSyncTime, batchSize}, 
            (rs, rowNum) -> {
                ChangeRecord record = new ChangeRecord();
                record.setId(rs.getLong("id"));
                record.setTableName(rs.getString("table_name"));
                record.setRecordId(rs.getString("record_id"));
                record.setOperationType(rs.getString("operation_type"));
                record.setChangeData(rs.getString("change_data"));
                record.setCreateTime(rs.getDate("create_time"));
                return record;
            });
    }
    /**
     * 执行增量同步
     */
    public SyncResult performSync(SyncConfig config) {
        SyncResult result = new SyncResult();
        result.setStartTime(new Date());
        try {
            // 1. 获取上次同步时间
            Date lastSyncTime = getLastSyncTime(config.getSyncKey());
            // 2. 批量获取待同步数据
            List<ChangeRecord> records = getPendingSyncRecords(
                lastSyncTime, config.getBatchSize());
            // 3. 执行同步
            for (ChangeRecord record : records) {
                boolean success = syncSingleRecord(record, config);
                if (success) {
                    markAsSynced(record.getId());
                    result.addSuccessRecord(record.getRecordId());
                } else {
                    result.addFailedRecord(record.getRecordId());
                }
            }
            // 4. 更新同步时间戳
            updateSyncTime(config.getSyncKey(), result.getStartTime());
            result.setEndTime(new Date());
            result.setSuccess(true);
        } catch (Exception e) {
            result.setSuccess(false);
            result.setErrorMessage(e.getMessage());
            log.error("增量同步失败", e);
        }
        return result;
    }
    /**
     * 同步单条记录
     */
    private boolean syncSingleRecord(ChangeRecord record, SyncConfig config) {
        try {
            switch (record.getOperationType()) {
                case "INSERT":
                    return insertToTarget(record, config);
                case "UPDATE":
                    return updateToTarget(record, config);
                case "DELETE":
                    return deleteFromTarget(record, config);
                default:
                    log.warn("未知操作类型: {}", record.getOperationType());
                    return false;
            }
        } catch (Exception e) {
            log.error("同步记录失败: {}", record.getRecordId(), e);
            return false;
        }
    }
    private Date getLastSyncTime(String syncKey) {
        String cacheKey = "sync:last_time:" + syncKey;
        Object value = redisTemplate.opsForValue().get(cacheKey);
        if (value != null) {
            return (Date) value;
        }
        // 默认从30天前开始
        return DateUtils.addDays(new Date(), -30);
    }
    private void updateSyncTime(String syncKey, Date syncTime) {
        String cacheKey = "sync:last_time:" + syncKey;
        redisTemplate.opsForValue().set(cacheKey, syncTime, 7, TimeUnit.DAYS);
    }
}

基于版本号的增量同步

/**
 * 数据版本管理
 */
@Component
public class VersionBasedSyncManager {
    @Autowired
    private DataSource dataSource;
    private final ConcurrentHashMap<String, AtomicLong> versionMap = new ConcurrentHashMap<>();
    /**
     * 获取数据版本
     */
    public Long getCurrentVersion(String tableName) {
        return versionMap.computeIfAbsent(tableName, k -> {
            try {
                String sql = "SELECT MAX(version) FROM " + tableName;
                return new AtomicLong(queryMaxVersion(sql));
            } catch (Exception e) {
                return new AtomicLong(0L);
            }
        }).get();
    }
    /**
     * 基于版本增量同步
     */
    public List<Map<String, Object>> syncByVersion(String tableName, 
                                                   Long lastVersion, 
                                                   int batchSize) {
        String sql = "SELECT * FROM " + tableName + 
                    " WHERE version > ? ORDER BY version ASC LIMIT ?";
        return jdbcTemplate.queryForList(sql, lastVersion, batchSize);
    }
    /**
     * 批量版本同步
     */
    @Transactional
    public BatchSyncResult batchVersionSync(String sourceTable, 
                                           String targetTable,
                                           Long lastSyncVersion) {
        BatchSyncResult result = new BatchSyncResult();
        while (true) {
            // 获取增量数据
            List<Map<String, Object>> records = syncByVersion(
                sourceTable, lastSyncVersion, 1000);
            if (records.isEmpty()) {
                break;
            }
            // 批量写入目标表
            int batchSize = batchInsertToTarget(targetTable, records);
            result.addProcessed(batchSize);
            // 更新版本号
            lastSyncVersion = records.get(records.size() - 1)
                                    .get("version") instanceof Long ? 
                (Long) records.get(records.size() - 1).get("version") : 
                Long.parseLong(records.get(records.size() - 1).get("version").toString());
        }
        result.setLastSyncVersion(lastSyncVersion);
        return result;
    }
}

基于CDC的增量同步（使用Debezium）

/**
 * Debezium CDC配置
 */
@Configuration
public class CdcSyncConfig {
    @Bean
    public DebeziumEngine<ChangeEvent<String, String>> debeziumEngine() {
        Properties props = new Properties();
        props.setProperty("name", "engine");
        props.setProperty("connector.class", 
            "io.debezium.connector.mysql.MySqlConnector");
        props.setProperty("offset.storage", 
            "org.apache.kafka.connect.storage.FileOffsetBackingStore");
        props.setProperty("offset.storage.file.filename", 
            "/tmp/offsets.dat");
        props.setProperty("offset.flush.interval.ms", "60000");
        props.setProperty("database.hostname", "localhost");
        props.setProperty("database.port", "3306");
        props.setProperty("database.user", "user");
        props.setProperty("database.password", "password");
        props.setProperty("database.server.id", "85744");
        props.setProperty("database.server.name", "my-app-connector");
        props.setProperty("database.history", 
            "io.debezium.relational.history.FileDatabaseHistory");
        props.setProperty("database.history.file.filename", 
            "/tmp/dbhistory.dat");
        props.setProperty("table.whitelist", "mydb.users,mydb.orders");
        props.setProperty("database.history.skip.unparseable.ddl", "true");
        props.setProperty("decimal.handling.mode", "string");
        return DebeziumEngine.create(ChangeEventFormat.of(Connect.class))
            .using(props)
            .notifying(this::handleChangeEvent)
            .build();
    }
    private void handleChangeEvent(ChangeEvent<String, String> event) {
        SourceRecord sourceRecord = (SourceRecord) event.value();
        Struct sourceStruct = (Struct) sourceRecord.value();
        // 解析变更事件
        String operation = sourceStruct.getString("op");
        Struct after = sourceStruct.getStruct("after");
        Struct before = sourceStruct.getStruct("before");
        // 根据操作类型处理
        switch (operation) {
            case "c": // 创建
                handleCreate(after);
                break;
            case "u": // 更新
                handleUpdate(before, after);
                break;
            case "d": // 删除
                handleDelete(before);
                break;
        }
    }
}

完整的增量同步框架

/**
 * 增量同步配置
 */
@Data
public class SyncConfig {
    private String syncKey;          // 同步标识
    private String sourceTable;      // 源表
    private String targetTable;      // 目标表
    private String primaryKey;       // 主键字段
    private String timestampField;   // 时间戳字段
    private String versionField;     // 版本字段
    private int batchSize = 1000;    // 批量大小
    private int retryCount = 3;      // 重试次数
    private long retryInterval = 1000L; // 重试间隔ms
}
/**
 * 增量同步执行器
 */
@Component
public class IncrementalSyncExecutor {
    @Autowired
    private ApplicationContext applicationContext;
    private final Map<String, SyncTask> runningTasks = new ConcurrentHashMap<>();
    /**
     * 启动同步任务
     */
    public void startSync(SyncConfig config) {
        SyncTask task = new SyncTask(config, applicationContext);
        runningTasks.put(config.getSyncKey(), task);
        Thread syncThread = new Thread(task, "sync-" + config.getSyncKey());
        syncThread.start();
    }
    /**
     * 停止同步任务
     */
    public void stopSync(String syncKey) {
        SyncTask task = runningTasks.get(syncKey);
        if (task != null) {
            task.stop();
            runningTasks.remove(syncKey);
        }
    }
    /**
     * 同步任务
     */
    private class SyncTask implements Runnable {
        private final SyncConfig config;
        private final IncrementalSyncService syncService;
        private volatile boolean running = true;
        public SyncTask(SyncConfig config, ApplicationContext context) {
            this.config = config;
            this.syncService = context.getBean(IncrementalSyncService.class);
        }
        @Override
        public void run() {
            while (running) {
                try {
                    SyncResult result = syncService.performSync(config);
                    if (!result.isSuccess()) {
                        log.error("同步失败: {}", result.getErrorMessage());
                        // 失败重试
                        handleSyncFailure(result);
                    }
                    // 间隔时间
                    Thread.sleep(config.getRetryInterval());
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                } catch (Exception e) {
                    log.error("同步异常", e);
                }
            }
        }
        public void stop() {
            this.running = false;
        }
    }
}

使用示例

@RestController
@RequestMapping("/sync")
public class SyncController {
    @Autowired
    private IncrementalSyncExecutor syncExecutor;
    @Autowired
    private IncrementalSyncService syncService;
    /**
     * 手动触发增量同步
     */
    @PostMapping("/manual")
    public ApiResponse manualSync(@RequestBody SyncRequest request) {
        SyncConfig config = new SyncConfig();
        config.setSyncKey(request.getSyncKey());
        config.setSourceTable(request.getSourceTable());
        config.setTargetTable(request.getTargetTable());
        config.setTimestampField("update_time");
        config.setBatchSize(1000);
        SyncResult result = syncService.performSync(config);
        return ApiResponse.success(result);
    }
    /**
     * 启动自动同步
     */
    @PostMapping("/start")
    public ApiResponse startAutoSync(@RequestBody SyncConfig config) {
        syncExecutor.startSync(config);
        return ApiResponse.success("同步任务已启动");
    }
    /**
     * 停止自动同步
     */
    @PostMapping("/stop/{syncKey}")
    public ApiResponse stopAutoSync(@PathVariable String syncKey) {
        syncExecutor.stopSync(syncKey);
        return ApiResponse.success("同步任务已停止");
    }
}

选择合适的同步策略：

1. 时间戳方式：简单易实现，适合数据量不大的场景
2. 版本号方式：支持乐观锁，适合并发场景
3. CDC方式：实时性高，适合大数据量场景
4. 混合方式：结合多种策略，应对复杂需求

关键考虑点：

• 数据一致性：使用事务保证同步的完整性
• 性能优化：批量处理、异步处理
• 错误处理：重试机制、死信队列
• 监控告警：同步延迟、失败记录监控

根据你的业务场景选择合适的同步策略,并做好异常处理和监控。