
## 一、数据同步概述

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

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

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

## 二、Canal原理解析

### 1. 工作原理

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

### 2. Binlog格式

**ROW格式的Binlog记录行变更：**

```sql
-- 查看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部署

```yaml
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配置

```ini
[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
```

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

### 4. Kafka消费者实现

```java
@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. 消费者实现

```java
@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. 同步策略

```java
@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：唯一键判断**

```java
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：数据库唯一索引**

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

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

## 七、数据同步监控

### 1. 延迟监控

```java
@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. 数据校验

```java
@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位点丢失

```java
// 定期保存消费位点
@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. 大表同步

```java
// 大表全量同步时使用分批处理
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. 做好数据校验和监控

---

*个人观点，仅供参考*

【架构实战】数据同步架构：Canal + MQ实时同步

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

实时性强：秒级延迟

解耦：Canal与消费者解耦

可扩展：可对接多个目标存储

幂等处理：防止重复消费

最佳实践：

研发部经理

人工智能

Canal实现MySQL数据实时同步至Redis、Elasticsearch、ClickHouse等存储系统，通过解析Binlog日志实现秒级延迟。采用Kafka消息队列解耦，确保数据一致性，支持全量/增量同步，提供幂等处理方案解决重复消费问题，包含延迟监控与数据校验机制，是分布式系统数据同步的可靠解决方案。

数据一致性

消息队列

数据同步

数据缓存

数据库

MySQL

Redis

Kafka

JSON

腾讯云数据库特惠，新用户首单0.5折起


2026采购季 | AI焕新·智启新局

tencentdb-catalog

tchousec

datainlong

message-queue-catalog

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云AI代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

功能1上新10个字符

功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符。

功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符。

功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符

功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符

功能4上新

文章&问答评论现已支持表情

全新交互，全新视觉，新增快捷键、悬浮工具栏、高亮块等功能并同时优化现有功能，全面提升创作效率和体验

社区富文本编辑器全新改版！诚邀体验～ 

精选全网热门MCP server，让你的AI更好用 🚀

💥开发者 MCP广场重磅上线！

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

【架构实战】数据同步架构：Canal + MQ实时同步-腾讯云开发者社区-腾讯云

【架构实战】数据同步架构：Canal + MQ实时同步

【架构实战】数据同步架构：Canal + MQ实时同步

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐