Java分布式数据同步API怎么实现

本文目录导读：

1. 核心实现方案
2. 完整方案设计
3. 高级特性实现
4. 方案对比与选择建议

实现Java分布式数据同步API需要根据具体场景选择合适的技术方案,我为你介绍几种主流实现方式，包括核心原理、代码示例和适用场景。

核心实现方案

基于消息队列的异步同步

这是最常用的松耦合方案,通过消息中间件实现数据变更的可靠传播。

架构设计

服务A -> 数据变更 -> 发送消息(发布) -> MQ Broker -> 消费者(服务B,C) -> 落库

代码实现

消息生产者（源服务）

@Component
public class DataChangeProducer {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    public void publishDataChange(DataChangeEvent event) {
        // 封装变更消息
        String message = JSON.toJSONString(event);
        // 发送到Kafka
        kafkaTemplate.send("data-sync-topic", 
            event.getEntityType(), 
            message
        ).addCallback(
            result -> {
                // 记录发送成功
                log.info("消息发送成功: {}", event.getEventId());
                // 更新消息状态
                redisTemplate.opsForValue().set(
                    "sync:msg:" + event.getEventId(), 
                    "SENT",
                    1, TimeUnit.DAYS
                );
            },
            failure -> {
                // 记录失败，后续补偿
                log.error("消息发送失败: {}", event.getEventId(), failure);
                saveToCompensationTable(event);
            }
        );
    }
    // 补偿表存储
    private void saveToCompensationTable(DataChangeEvent event) {
        // 存入本地数据库补偿表
    }
}

消息消费者（目标服务）

@Component
@Slf4j
public class DataSyncConsumer {
    @Autowired
    private DataSyncService syncService;
    @Autowired
    private RedissonClient redissonClient;
    @KafkaListener(topics = "data-sync-topic", 
                  groupId = "data-sync-group")
    public void consume(ConsumerRecord<String, String> record) {
        DataChangeEvent event = JSON.parseObject(record.value(), DataChangeEvent.class);
        String eventId = event.getEventId();
        // 分布式幂等锁
        RLock lock = redissonClient.getLock("sync:lock:" + eventId);
        try {
            // 尝试加锁，防止重复消费
            if (lock.tryLock(5, 10, TimeUnit.SECONDS)) {
                // 幂等性检查
                if (syncService.isEventProcessed(eventId)) {
                    log.info("事件已处理，跳过: {}", eventId);
                    return;
                }
                // 执行数据同步
                syncService.syncData(event);
                // 标记已处理
                syncService.markEventProcessed(eventId);
            }
        } catch (Exception e) {
            log.error("同步处理失败, eventId: {}", eventId, e);
            // 记录失败，由补偿机制处理
            throw new RuntimeException("同步失败", e);
        } finally {
            lock.unlock();
        }
    }
}

基于数据库Binlog的CDC方案

实时捕获数据库变更,实现不同数据源之间的同步，适合异构数据库同步。

使用Canal组件

@Component
public class BinlogSyncService {
    @Autowired
    private CanalConnector canalConnector;
    @Autowired
    private DataMappingService mappingService;
    @PostConstruct
    public void startSync() {
        Thread syncThread = new Thread(() -> {
            while (true) {
                try {
                    syncFromBinlog();
                } catch (Exception e) {
                    log.error("Binlog同步异常", e);
                    sleep(1000);
                }
            }
        });
        syncThread.setDaemon(true);
        syncThread.start();
    }
    private void syncFromBinlog() {
        // 连接Canal Server
        canalConnector.connect();
        canalConnector.subscribe(".*\\..*");
        canalConnector.rollback();
        while (true) {
            Message message = canalConnector.getWithoutAck(100, 1000);
            long batchId = message.getId();
            if (batchId == -1 || message.getEntries().isEmpty()) {
                continue;
            }
            try {
                // 解析并处理Binlog事件
                for (CanalEntry.Entry entry : message.getEntries()) {
                    if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                        parseAndSync(entry);
                    }
                }
                // 确认已处理
                canalConnector.ack(batchId);
            } catch (Exception e) {
                log.error("处理Binlog失败, batchId: {}", batchId, e);
                canalConnector.rollback(batchId);
            }
        }
    }
    private void parseAndSync(CanalEntry.Entry entry) {
        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
        CanalEntry.EventType eventType = rowChange.getEventType();
        for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
            // 转换为目标数据格式
            SyncData syncData = mappingService.mapData(
                entry.getHeader().getTableName(),
                eventType,
                rowData
            );
            // 执行同步
            syncToTarget(syncData);
        }
    }
    private void syncToTarget(SyncData syncData) {
        // 根据事件类型处理
        switch (syncData.getEventType()) {
            case INSERT:
                // 写入目标数据库
                break;
            case UPDATE:
                // 更新目标数据库
                break;
            case DELETE:
                // 删除目标数据库记录
                break;
        }
    }
}

分布式锁+数据库记录同步

适合少量数据、强一致性要求的场景，通过分布式锁保证同步顺序。

@Service
public class DistributedDataSyncService {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    @Autowired
    private ZooKeeperClient zkClient;
    @Autowired
    private JdbcTemplate jdbcTemplate;
    // 同步版本号
    private static final String SYNC_VERSION_KEY = "data:sync:version:";
    /**
     * 同步数据（使用ZooKeeper分布式锁）
     */
    public boolean syncDataWithZK(SyncRequest request) {
        // 创建ZooKeeper事务锁
        InterProcessMutex lock = new InterProcessMutex(
            zkClient.getClient(), 
            "/data-sync/" + request.getEntityType()
        );
        try {
            // 加锁，最多等待10秒
            if (!lock.acquire(10, TimeUnit.SECONDS)) {
                throw new RuntimeException("获取同步锁超时");
            }
            // 版本检查
            String versionKey = SYNC_VERSION_KEY + request.getEntityId();
            Long currentVersion = redisTemplate.opsForValue()
                .increment(versionKey, 0);
            if (request.getVersion() <= currentVersion) {
                log.warn("数据版本过低，忽略同步");
                return false;
            }
            // 执行数据同步逻辑
            executeSync(request);
            // 更新版本号
            redisTemplate.opsForValue()
                .set(versionKey, request.getVersion());
            return true;
        } catch (Exception e) {
            log.error("同步失败", e);
            throw new RuntimeException("同步失败", e);
        } finally {
            try {
                lock.release();
            } catch (Exception e) {
                log.error("释放锁失败", e);
            }
        }
    }
    private void executeSync(SyncRequest request) {
        // 使用数据库事务保证一致性
        TransactionTemplate transactionTemplate = new TransactionTemplate();
        transactionTemplate.execute(status -> {
            try {
                // 1. 查询源数据
                Map<String, Object> sourceData = querySourceData(request);
                // 2. 转换数据格式
                Map<String, Object> targetData = transformData(sourceData);
                // 3. 写入目标数据库
                upsertTargetData(targetData);
                return true;
            } catch (Exception e) {
                status.setRollbackOnly();
                throw e;
            }
        });
    }
}

HTTP/GRPC双向同步

适合服务间直接通信,实时性要求高的场景。

// 使用Spring Cloud OpenFeign实现HTTP同步
@FeignClient(name = "data-sync-client", 
             url = "${target.service.url}")
public interface DataSyncApi {
    @PostMapping("/api/sync/data")
    SyncResponse syncData(@RequestBody SyncRequest request);
    @PostMapping("/api/sync/batch")
    SyncResponse syncBatchData(@RequestBody List<SyncRequest> requests);
}
// 服务端API实现
@RestController
@RequestMapping("/api/sync")
public class DataSyncController {
    @PostMapping("/data")
    public SyncResponse receiveSync(@RequestBody SyncRequest request) {
        // 验证请求签名
        if (!validateSignature(request)) {
            return SyncResponse.fail("签名验证失败");
        }
        // 分布式ID去重
        if (idempotentService.isDuplicate(request.getRequestId())) {
            return SyncResponse.success("请求已处理");
        }
        // 异步处理同步请求
        asyncSyncService.submitSyncTask(request);
        return SyncResponse.success();
    }
    @PostMapping("/batch")
    public SyncResponse receiveBatchSync(@RequestBody List<SyncRequest> requests) {
        // 批量处理，提高效率
        for (SyncRequest request : requests) {
            if (!validateSignature(request)) {
                return SyncResponse.fail("签名验证失败: " + request.getRequestId());
            }
        }
        // 批量提交
        asyncSyncService.submitBatchSyncTask(requests);
        return SyncResponse.success();
    }
}

完整方案设计

项目依赖配置 (pom.xml)

<dependencies>
    <!-- Kafka -->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <!-- Canal -->
    <dependency>
        <groupId>com.alibaba.otter</groupId>
        <artifactId>canal.client</artifactId>
        <version>1.1.6</version>
    </dependency>
    <!-- ZooKeeper -->
    <dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-recipes</artifactId>
        <version>5.4.0</version>
    </dependency>
    <!-- Redisson -->
    <dependency>
        <groupId>org.redisson</groupId>
        <artifactId>redisson-spring-boot-starter</artifactId>
        <version>3.22.1</version>
    </dependency>
    <!-- Spring Retry -->
    <dependency>
        <groupId>org.springframework.retry</groupId>
        <artifactId>spring-retry</artifactId>
    </dependency>
</dependencies>

配置示例

# application.yml
spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      retries: 3
      acks: all
    consumer:
      auto-offset-reset: earliest
      enable-auto-commit: false
  redis:
    host: localhost
    port: 6379
canal:
  server: localhost:11111
  instance: example
  batch-size: 1000
zookeeper:
  connect-string: localhost:2181
  session-timeout: 10000
  connection-timeout: 5000

高级特性实现

数据一致性检查

@Component
public class ConsistencyChecker {
    @Scheduled(cron = "0 0 2 * * ?") // 每天凌晨2点运行
    public void checkConsistency() {
        // 获取所有需要检查的实体类型
        List<String> entityTypes = getEntityTypes();
        for (String entityType : entityTypes) {
            // 分页比对源和目标数据
            int page = 1;
            int pageSize = 1000;
            while (true) {
                List<String> ids = queryAllIds(entityType, page, pageSize);
                if (ids.isEmpty()) break;
                // 并行检查
                ids.parallelStream().forEach(id -> {
                    boolean consistent = checkSingleRecord(entityType, id);
                    if (!consistent) {
                        // 记录不一致，触发修复
                        repairData(entityType, id);
                    }
                });
                page++;
            }
        }
    }
    private boolean checkSingleRecord(String entityType, String entityId) {
        // 获取源端数据
        Map<String, Object> sourceData = getSourceData(entityType, entityId);
        // 获取目标端数据
        Map<String, Object> targetData = getTargetData(entityType, entityId);
        // 对比关键字段
        return compareData(sourceData, targetData);
    }
}

自动补偿机制

@Component
public class CompensationService {
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    @Retryable(value = Exception.class, 
               maxAttempts = 3, 
               backoff = @Backoff(delay = 2000, multiplier = 2))
    public void compensateFailedSync(CompensationTask task) {
        // 1. 检查当前状态
        String statusKey = "sync:status:" + task.getEventId();
        String status = (String) redisTemplate.opsForValue().get(statusKey);
        if ("SUCCESS".equals(status)) {
            return; // 已成功，不用再补偿
        }
        // 2. 重新执行同步
        SyncRequest request = convertToRequest(task);
        boolean result = syncService.syncData(request);
        if (result) {
            // 3. 标记成功
            redisTemplate.opsForValue()
                .set(statusKey, "SUCCESS", 7, TimeUnit.DAYS);
            // 4. 清理补偿记录
            clearCompensationTask(task.getId());
        } else {
            // 5. 如果失败，记录并告警
            log.error("补偿失败: {}", task);
            alertService.sendAlert("数据同步补偿失败", task);
        }
    }
}

方案对比与选择建议

选择建议：

• 数据实时性要求高 → HTTP/GRPC直连方案
• 系统解耦、异步处理 → 消息队列方案
• 异构数据库同步 → Binlog CDC方案
• 少量关键数据 → 分布式锁+记录方案

在实际项目中,通常组合多种方案使用，例如核心数据用HTTP直连（强一致性），非核心数据用消息队列（高性能）。