## 一、数据一致性的挑战

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

```
MySQL（主数据）
    ├── Redis（缓存）
    ├── Elasticsearch（搜索）
    ├── ClickHouse（分析）
    └── 其他服务的数据库
```

**核心问题：** 当MySQL数据变更时，如何保证其他存储的数据也同步更新？

**常见方案：**
1. **双写**：应用层同时写多个存储
2. **Binlog同步**：监听MySQL Binlog，异步同步
3. **消息队列**：写完MySQL后发消息，消费者同步

## 二、双写方案

### 1. 实现方式

```java
@Service
@Transactional
public class ProductService {
    
    @Autowired
    private ProductMapper productMapper;
    
    @Autowired
    private RedisTemplate redisTemplate;
    
    @Autowired
    private ElasticsearchClient esClient;
    
    public void updateProduct(Product product) {
        // 1. 更新MySQL
        productMapper.updateById(product);
        
        // 2. 更新Redis
        redisTemplate.opsForValue().set(
            "product:" + product.getId(), product);
        
        // 3. 更新ES
        esClient.index(i -> i
            .index("products")
            .id(product.getId().toString())
            .document(product));
    }
}
```

### 2. 双写的问题

```
问题1：原子性
MySQL更新成功，Redis更新失败 → 数据不一致

问题2：性能
同步写多个存储，响应时间增加

问题3：耦合
业务代码与存储细节耦合
```

## 三、Binlog同步方案（Canal）

### 1. Canal原理

```
MySQL主库
    │
    │ Binlog（ROW格式）
    ▼
Canal Server（伪装成MySQL从库）
    │
    │ 解析Binlog事件
    ▼
Canal Client
    │
    ├──► Redis
    ├──► Elasticsearch
    └──► 其他存储
```

### 2. Canal部署

**Docker方式：**

```yaml
version: '3'
services:
  canal-server:
    image: canal/canal-server:v1.1.6
    container_name: canal-server
    ports:
      - "11111:11111"
    environment:
      canal.instance.master.address: mysql:3306
      canal.instance.dbUsername: canal
      canal.instance.dbPassword: canal
      canal.instance.filter.regex: order_db\..*
    volumes:
      - ./canal-logs:/home/admin/canal-server/logs
```

**MySQL配置：**

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

### 3. Canal Client实现

**依赖：**

```xml
<dependency>
    <groupId>com.alibaba.otter</groupId>
    <artifactId>canal.client</artifactId>
    <version>1.1.6</version>
</dependency>
```

**消费Binlog：**

```java
@Component
@Slf4j
public class CanalClient {
    
    @Autowired
    private RedisTemplate redisTemplate;
    
    @Autowired
    private ElasticsearchClient esClient;
    
    private CanalConnector connector;
    
    @PostConstruct
    public void start() {
        connector = CanalConnectors.newSingleConnector(
            new InetSocketAddress("canal-server", 11111),
            "example",
            "",
            ""
        );
        
        connector.connect();
        connector.subscribe("order_db\..*");
        connector.rollback();
        
        // 启动消费线程
        Thread thread = new Thread(this::consume);
        thread.setDaemon(true);
        thread.start();
    }
    
    private void consume() {
        while (true) {
            try {
                Message message = connector.getWithoutAck(100);
                long batchId = message.getId();
                
                if (batchId == -1 || message.getEntries().isEmpty()) {
                    Thread.sleep(1000);
                    continue;
                }
                
                processEntries(message.getEntries());
                connector.ack(batchId);
                
            } catch (Exception e) {
                log.error("Canal消费异常", e);
                connector.rollback();
            }
        }
    }
    
    private void processEntries(List<Entry> entries) {
        for (Entry entry : entries) {
            if (entry.getEntryType() != EntryType.ROWDATA) {
                continue;
            }
            
            RowChange rowChange;
            try {
                rowChange = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                log.error("解析Binlog失败", e);
                continue;
            }
            
            String tableName = entry.getHeader().getTableName();
            EventType eventType = rowChange.getEventType();
            
            for (RowData rowData : rowChange.getRowDatasList()) {
                if (eventType == EventType.INSERT || eventType == EventType.UPDATE) {
                    handleUpsert(tableName, rowData.getAfterColumnsList());
                } else if (eventType == EventType.DELETE) {
                    handleDelete(tableName, rowData.getBeforeColumnsList());
                }
            }
        }
    }
    
    private void handleUpsert(String tableName, List<Column> columns) {
        Map<String, String> data = columns.stream()
            .collect(Collectors.toMap(Column::getName, Column::getValue));
        
        if ("product".equals(tableName)) {
            Long id = Long.parseLong(data.get("id"));
            
            // 更新Redis
            redisTemplate.opsForValue().set(
                "product:" + id, data, 30, TimeUnit.MINUTES);
            
            // 更新ES
            esClient.index(i -> i
                .index("products")
                .id(id.toString())
                .document(data));
            
            log.info("同步商品数据: {}", id);
        }
    }
    
    private void handleDelete(String tableName, List<Column> columns) {
        Map<String, String> data = columns.stream()
            .collect(Collectors.toMap(Column::getName, Column::getValue));
        
        if ("product".equals(tableName)) {
            Long id = Long.parseLong(data.get("id"));
            
            // 删除Redis
            redisTemplate.delete("product:" + id);
            
            // 删除ES
            esClient.delete(d -> d
                .index("products")
                .id(id.toString()));
            
            log.info("删除商品数据: {}", id);
        }
    }
}
```

### 4. Canal + MQ模式

```java
@Component
public class CanalToMqBridge {
    
    @Autowired
    private KafkaTemplate kafkaTemplate;
    
    private void processEntries(List<Entry> entries) {
        for (Entry entry : entries) {
            // 将Binlog事件转换为消息
            BinlogEvent event = convertToEvent(entry);
            
            // 发送到Kafka
            kafkaTemplate.send("binlog-events", event);
        }
    }
}

@KafkaListener(topics = "binlog-events")
public class BinlogEventConsumer {
    
    public void consume(BinlogEvent event) {
        // 根据表名和操作类型处理
        switch (event.getTableName()) {
            case "product":
                syncProductToRedis(event);
                syncProductToEs(event);
                break;
            case "order":
                syncOrderToEs(event);
                break;
        }
    }
}
```

## 四、消息队列方案

### 1. 实现方式

```java
@Service
@Transactional
public class ProductService {
    
    @Autowired
    private ProductMapper productMapper;
    
    @Autowired
    private KafkaTemplate kafkaTemplate;
    
    public void updateProduct(Product product) {
        // 1. 更新MySQL
        productMapper.updateById(product);
        
        // 2. 发送消息（事务提交后发送）
        TransactionSynchronizationManager.registerSynchronization(
            new TransactionSynchronizationAdapter() {
                @Override
                public void afterCommit() {
                    kafkaTemplate.send("product:update", product);
                }
            }
        );
    }
}

@KafkaListener(topics = "product:update")
public class ProductSyncConsumer {
    
    @Autowired
    private RedisTemplate redisTemplate;
    
    @Autowired
    private ElasticsearchClient esClient;
    
    public void consume(Product product) {
        // 同步到Redis
        redisTemplate.opsForValue().set(
            "product:" + product.getId(), product);
        
        // 同步到ES
        esClient.index(i -> i
            .index("products")
            .id(product.getId().toString())
            .document(product));
    }
}
```

### 2. 事务消息（RocketMQ）

```java
@Service
public class ProductService {
    
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    
    public void updateProduct(Product product) {
        // 发送事务消息
        rocketMQTemplate.sendMessageInTransaction(
            "product-update-topic",
            MessageBuilder.withPayload(product).build(),
            product
        );
    }
}

@RocketMQTransactionListener
public class ProductTransactionListener implements RocketMQLocalTransactionListener {
    
    @Autowired
    private ProductMapper productMapper;
    
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(
            Message msg, Object arg) {
        try {
            Product product = (Product) arg;
            productMapper.updateById(product);
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }
    
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        // 检查本地事务状态
        Product product = (Product) msg.getPayload();
        Product dbProduct = productMapper.selectById(product.getId());
        
        if (dbProduct != null && dbProduct.getVersion().equals(product.getVersion())) {
            return RocketMQLocalTransactionState.COMMIT;
        }
        
        return RocketMQLocalTransactionState.ROLLBACK;
    }
}
```

## 五、方案对比

| 维度 | 双写 | Binlog同步 | 消息队列 |
|------|------|------------|----------|
| 实时性 | 强 | 准实时 | 准实时 |
| 一致性 | 弱（可能失败） | 强 | 中（需幂等） |
| 耦合度 | 高 | 低 | 中 |
| 复杂度 | 低 | 高 | 中 |
| 性能影响 | 大 | 小 | 小 |
| 适用场景 | 简单场景 | 数据同步 | 业务解耦 |

## 六、总结

数据一致性是分布式系统的核心挑战：

- **双写**：简单但可靠性差
- **Binlog同步**：可靠但复杂，适合数据同步
- **消息队列**：解耦但需要幂等处理

**推荐方案：**
1. 缓存同步：Binlog + Canal
2. 搜索同步：Binlog + Canal + ES
3. 业务解耦：消息队列 + 幂等消费

**思考题**：你们系统如何保证多存储之间的数据一致性？

【架构实战】数据一致性方案：Binlog同步 vs 消息队列

数据一致性方案：Binlog同步 vs 消息队列

研发部经理

编程语言

分布式系统数据一致性解决方案对比：双写方案简单但可靠性差，Binlog同步可靠但复杂，消息队列解耦需幂等处理。推荐缓存同步用Binlog+Canal，搜索同步用Binlog+Canal+ES，业务解耦用消息队列+幂等消费。三种方案各具优势，需根据业务场景选择最佳实现方式。

数据一致性

消息队列

数据同步

数据库

MySQL

Redis

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

ckafka

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


message-queue-catalog

datainlong

tencentdb-catalog

tchousec

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

社区新版编辑器体验调研

【架构实战】数据一致性方案：Binlog同步 vs 消息队列-腾讯云开发者社区-腾讯云

【架构实战】数据一致性方案：Binlog同步 vs 消息队列

【架构实战】数据一致性方案：Binlog同步 vs 消息队列

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐