一、数据同步概述

在分布式系统中，数据通常存储在多个地方：

MySQL（业务数据库）
    │
    ├──► Redis（热点数据缓存）
    ├──► Elasticsearch（搜索引擎）
    ├──► ClickHouse（数据分析）
    └──► Kafka（消息队列）

核心问题： 如何保证多数据源之间的数据一致性？

二、Canal原理解析

1. 工作原理

MySQL主库
    │
    │ 开启Binlog（ROW格式）
    ▼
Canal Server（伪装成从库）
    │
    │ 解析Binlog
    ▼
Canal Client
    │
    │ 发送消息
    ▼
消息队列/目标存储

2. Binlog格式

ROW格式的Binlog记录行变更：

-- 查看Binlog格式
SHOW VARIABLES LIKE 'binlog_format';
-- 结果：ROW

-- 查看当前Binlog文件
SHOW MASTER STATUS;
-- 结果：mysql-bin.000001, 位置154

三、Canal + Kafka实时同步

1. 架构设计

MySQL ──► Canal Server ──► Kafka ──► 消费者 ──► 多个目标存储
                              │
                              ├──► Redis
                              ├──► Elasticsearch
                              └──► ClickHouse

2. Canal Server部署

version: '3'
services:
  canal:
    image: canal/canal-server:v1.1.6
    container_name: canal-server
    ports:
      - "11110:11110"  # TCP
      - "11111:11111"  # metrics
    environment:
      canal.instance.mysql.slaveId: 1234
      canal.instance.master.address: mysql:3306
      canal.instance.dbUsername: canal
      canal.instance.dbPassword: canal
      canal.instance.filter.regex: order_db\\..*
      canal.mq.topic: binlog-data
      canal.mq.dynamicTopic: true
    volumes:
      - ./canal-logs:/home/admin/canal-server/logs

3. MySQL配置

[mysqld]
server-id = 1
log-bin = mysql-bin
binlog-format = ROW
expire-logs-days = 7
max-binlog-size = 100M

# 需要同步的数据库
binlog-do-db = order_db

# 不需要同步的数据库
binlog-ignore-db = mysql
binlog-ignore-db = information_schema

-- 创建Canal用户
CREATE USER 'canal'@'%' IDENTIFIED BY 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;

4. Kafka消费者实现

@Component
@Slf4j
public class BinlogKafkaConsumer {
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private ElasticsearchClient esClient;
    
    @KafkaListener(topics = "binlog-data", groupId = "binlog-consumer")
    public void consume(String message) {
        try {
            BinlogMessage binlogMessage = JSON.parseObject(message, BinlogMessage.class);
            
            String tableName = binlogMessage.getTableName();
            EventType eventType = binlogMessage.getEventType();
            List<Map<String, String>> data = binlogMessage.getData();
            
            for (Map<String, String> row : data) {
                switch (tableName) {
                    case "product":
                        handleProductChange(eventType, row);
                        break;
                    case "order":
                        handleOrderChange(eventType, row);
                        break;
                    case "user":
                        handleUserChange(eventType, row);
                        break;
                }
            }
            
        } catch (Exception e) {
            log.error("处理Binlog消息失败: {}", message, e);
            // 发送告警
            alertService.alert("Binlog消费失败", e.getMessage());
        }
    }
    
    private void handleProductChange(EventType eventType, Map<String, String> data) {
        Long productId = Long.parseLong(data.get("id"));
        String key = "product:" + productId;
        
        switch (eventType) {
            case INSERT:
            case UPDATE:
                // 写入Redis
                redisTemplate.opsForValue().set(key, data, 30, TimeUnit.MINUTES);
                
                // 写入ES
                try {
                    esClient.index(i -> i
                        .index("products")
                        .id(productId.toString())
                        .document(data));
                } catch (IOException e) {
                    log.error("ES写入失败", e);
                }
                
                log.info("同步商品数据: {} - {}", eventType, productId);
                break;
                
            case DELETE:
                // 删除Redis
                redisTemplate.delete(key);
                
                // 删除ES
                try {
                    esClient.delete(d -> d
                        .index("products")
                        .id(productId.toString()));
                } catch (IOException e) {
                    log.error("ES删除失败", e);
                }
                
                log.info("删除商品数据: {}", productId);
                break;
        }
    }
}

四、Canal + ClickHouse同步

1. 消费者实现

@Component
@Slf4j
public class ClickHouseSink {
    
    @Autowired
    private JdbcTemplate clickHouseTemplate;
    
    public void sinkOrderData(List<Map<String, String>> dataList) {
        String sql = """
            INSERT INTO order_olap (order_id, order_no, user_id, shop_id, 
                                   order_amount, pay_amount, order_status, order_time)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            """;
        
        List<Object[]> batchArgs = dataList.stream()
            .map(data -> new Object[]{
                Long.parseLong(data.get("id")),
                data.get("order_no"),
                Long.parseLong(data.get("user_id")),
                Long.parseLong(data.get("shop_id")),
                new BigDecimal(data.get("order_amount")),
                new BigDecimal(data.get("pay_amount")),
                Integer.parseInt(data.get("order_status")),
                LocalDateTime.parse(data.get("order_time"))
            })
            .collect(Collectors.toList());
        
        clickHouseTemplate.batchUpdate(sql, batchArgs);
        log.info("同步订单数据到ClickHouse: {}条", dataList.size());
    }
}

五、Canal + Redis同步

1. 同步策略

@Component
@Slf4j
public class RedisSyncService {
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    // 全量同步
    public void fullSync(String tableName) {
        log.info("开始全量同步: {}", tableName);
        
        // 读取全量数据
        List<Map<String, Object>> allData = loadFullData(tableName);
        
        // 批量写入Redis
        for (Map<String, Object> row : allData) {
            String key = buildKey(tableName, row);
            redisTemplate.opsForValue().set(key, row);
        }
        
        log.info("全量同步完成: {} - {}条", tableName, allData.size());
    }
    
    // 增量同步
    public void incrementalSync(String tableName, List<Map<String, String>> dataList) {
        for (Map<String, String> row : dataList) {
            String key = buildKey(tableName, row);
            redisTemplate.opsForValue().set(key, row);
        }
        
        log.info("增量同步完成: {} - {}条", tableName, dataList.size());
    }
    
    // 删除同步
    public void deleteSync(String tableName, Map<String, String> data) {
        String key = buildKey(tableName, data);
        redisTemplate.delete(key);
        
        log.info("删除同步完成: {}", key);
    }
    
    private String buildKey(String tableName, Map<?, ?> data) {
        Long id = Long.parseLong(data.get("id").toString());
        return tableName + ":" + id;
    }
}

六、消息幂等处理

1. 问题分析

Canal发送消息 → Kafka → 消费者
     │                   │
     └───── 网络超时 ─────┘
                  │
           消息重复发送
                  │
           数据重复写入

2. 幂等方案

方案1：唯一键判断

public void processWithIdempotent(String tableName, Map<String, String> data) {
    Long id = Long.parseLong(data.get("id"));
    String key = tableName + ":processed:" + id;
    
    // 使用Redis实现分布式锁
    Boolean success = redisTemplate.opsForValue()
        .setIfAbsent(key, "1", 24, TimeUnit.HOURS);
    
    if (!success) {
        log.info("消息已处理过，跳过: {}", id);
        return;
    }
    
    try {
        // 执行业务逻辑
        doProcess(tableName, data);
    } catch (Exception e) {
        // 处理失败，删除幂等标记
        redisTemplate.delete(key);
        throw e;
    }
}

方案2：数据库唯一索引

-- 在目标表中创建唯一索引
CREATE UNIQUE INDEX idx_sync_id ON order_sync (sync_id);

-- 写入时使用INSERT IGNORE
INSERT IGNORE INTO order_sync (sync_id, order_data) VALUES (?, ?);

七、数据同步监控

1. 延迟监控

@Component
@Slf4j
public class SyncMonitor {
    
    @Autowired
    private CanalClient canalClient;
    
    @Scheduled(fixedRate = 60000)
    public void checkSyncDelay() {
        // 获取Canal消费位点
        long consumerPosition = canalClient.getConsumerPosition();
        
        // 获取MySQL当前位点
        long masterPosition = getMasterPosition();
        
        long delay = masterPosition - consumerPosition;
        
        log.info("同步延迟: {} bytes", delay);
        
        if (delay > 100 * 1024 * 1024) { // 100MB
            alertService.alert("同步延迟过大", "延迟: " + delay + " bytes");
        }
    }
}

2. 数据校验

@Service
public class DataValidator {
    
    @Autowired
    private ProductMapper mysqlProductMapper;
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Scheduled(cron = "0 0 2 * * ?") // 每天凌晨2点
    public void validateProductData() {
        log.info("开始数据校验");
        
        // 抽样校验100条数据
        List<Product> mysqlProducts = mysqlProductMapper.selectRandomProducts(100);
        
        int mismatchCount = 0;
        for (Product p : mysqlProducts) {
            String key = "product:" + p.getId();
            Product redisProduct = (Product) redisTemplate.opsForValue().get(key);
            
            if (!Objects.equals(p.getName(), redisProduct.getName()) ||
                !Objects.equals(p.getPrice(), redisProduct.getPrice())) {
                mismatchCount++;
                log.error("数据不一致: id={}", p.getId());
                
                // 修复不一致数据
                redisTemplate.opsForValue().set(key, p);
            }
        }
        
        log.info("数据校验完成，不一致数量: {}", mismatchCount);
        
        if (mismatchCount > 10) {
            alertService.alert("数据不一致数量过多", "不一致: " + mismatchCount);
        }
    }
}

八、常见问题处理

1. Binlog位点丢失

// 定期保存消费位点
@Scheduled(fixedRate = 300000) // 5分钟
public void saveConsumerPosition() {
    long position = canalClient.getConsumerPosition();
    
    // 保存到数据库
    consumerOffsetMapper.updateOffset(position);
    
    log.info("保存消费位点: {}", position);
}

// 启动时恢复位点
@PostConstruct
public void restorePosition() {
    Long lastPosition = consumerOffsetMapper.getLastOffset();
    if (lastPosition != null) {
        canalClient.seek(lastPosition);
    }
}

2. 大表同步

// 大表全量同步时使用分批处理
public void fullSyncWithBatch(String tableName) {
    long totalCount = mysqlMapper.getCount(tableName);
    int batchSize = 10000;
    int totalPages = (int) Math.ceil((double) totalCount / batchSize);
    
    for (int page = 0; page < totalPages; page++) {
        List<Map<String, Object>> batch = mysqlMapper.selectPage(
            tableName, page * batchSize, batchSize);
        
        // 同步到目标
        syncToTarget(batch);
        
        // 每批之间暂停，避免对源库压力太大
        if (page % 10 == 0) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

九、总结

Canal + MQ是数据同步的经典方案：

• 实时性强：秒级延迟
• 解耦：Canal与消费者解耦
• 可扩展：可对接多个目标存储
• 幂等处理：防止重复消费

最佳实践：

1. 开启Binlog ROW格式
2. 使用Kafka解耦
3. 实现幂等处理
4. 做好数据校验和监控

---

个人观点，仅供参考