常见的几种数据同步方案

常见的几种数据同步方案

引用

CSDN

1.

https://blog.csdn.net/zhousenshan/article/details/136432210

在当今数据驱动的时代，确保不同系统间的数据一致性、实时性和可靠性至关重要。本文将深入探讨五种常见的数据同步方案，包括数据库主从复制、ETL工具数据迁移、基于数据库触发器的同步、手工数据脚本同步以及实时数据同步方案（使用消息队列）。通过对比分析各种方案的特点和适用场景，帮助读者选择最适合的技术方案。

数据库主从复制

数据库主从复制是一种常见的数据同步方案，其中主数据库将其变更操作传播到一个或多个从数据库。

配置步骤

1. 确保主从数据库版本一致
2. 配置主数据库

   server-id = 1
   log_bin = /var/log/mysql/mysql-bin.log
   binlog_do_db = your_database_name
   

3. 创建复制用户

   create user 'replication_user'@'%' identified by 'replication_password';
   grant replication slave on *.* to 'replication_user'@'%';
   flush privileges;
   

4. 获取主数据库的二进制日志位置

   SHOW MASTER STATUS;
   

5. 配置从数据库

   server-id = 2
   

6. 连接从数据库到主数据库

   change master to
   master_host = 'master_host',
   master_user = 'replication_user',
   master_password = 'replication_password',
   master_log_file = 'master_log_file',
   master_log_pos = master_log_pos;
   

7. 启动从数据库的复制进程

   START SLAVE;
   

8. 验证复制状态

   SHOW SLAVE STATUS\G
   

优势：

• 提高读取性能
• 分担主数据库负载
• 提供容灾和备份机制

局限性：

• 存在复制延迟
• 主库单点故障可能影响整个系统
• 不适用于写入密集型应用

ETL工具数据迁移

ETL（Extract, Transform, Load）工具广泛用于不同数据存储系统之间的数据迁移、整合和同步。常见的ETL工具有：

以Apache NiFi为例，其单机架构主要包括：

• Web Server
• Flow Controller
• Extensions
• FlowFile Repository
• Content Repository
• Provenance Repository

触发器增量数据同步

触发器增量数据同步通过在源数据库中设置触发器，实现实时数据同步。例如：

DELIMITER //
CREATE TRIGGER sync_order_to_history
AFTER INSERT ON tb_order
FOR EACH ROW
BEGIN
    INSERT INTO tb_order_his (
        order_id, customer_id, order_date, product_id, quantity, 
        total_price, status, shipping_address, payment_method, 
        coupon_code, create_time, update_time
    )
    VALUES (
        NEW.order_id, NEW.customer_id, NEW.order_date, NEW.product_id, NEW.quantity, 
        NEW.total_price, NEW.status, NEW.shipping_address, NEW.payment_method, 
        NEW.coupon_code, NEW.create_time, NEW.update_time
    );
END;
//
DELIMITER ;

优点：

• 实时性
• 简化操作
• 确保数据一致性

缺点：

• 性能影响
• 复杂性
• 并发控制

手工脚本同步

手工脚本同步通过编写SQL脚本实现数据同步，适用于小规模数据的同步。例如：

INSERT INTO tb_order_his (
    order_id, customer_id, order_date, product_id, quantity,
    total_price, status, shipping_address, payment_method,
    coupon_code, create_time, update_time
)
SELECT
    order_id, customer_id, order_date, product_id, quantity,
    total_price, status, shipping_address, payment_method,
    coupon_code, create_time, update_time
FROM tb_order;

实时数据同步方案（使用消息队列）

实时数据同步方案通过消息队列实现数据变更的实时同步。例如，使用Debezium和Kafka：

1. 配置MySQL连接信息

   database.hostname=mysql-host
   database.port=3306
   database.user=mysql-user
   database.password=mysql-password
   connector.class=io.debezium.connector.mysql.MySqlConnector
   tasks.max=1
   database.server.id=1
   database.server.name=my-app-connector
   database.whitelist=mydatabase
   

2. 启动Debezium连接器

   debezium-connector-mysql my-connector.properties
   

3. 创建Kafka主题

   kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic my-topic
   

4. Java代码示例

   import org.apache.kafka.clients.consumer.Consumer;
   import org.apache.kafka.clients.consumer.ConsumerConfig;
   import org.apache.kafka.clients.consumer.ConsumerRecords;
   import org.apache.kafka.clients.consumer.KafkaConsumer;
   import org.apache.kafka.common.serialization.StringDeserializer;
   import java.time.Duration;
   import java.util.Collections;
   import java.util.Properties;
   
   public class ClickHouseDataConsumer {
       private static final String KAFKA_BOOTSTRAP_SERVERS = "localhost:9092";
       private static final String KAFKA_TOPIC = "my-topic";
       private static final String CLICKHOUSE_URL = "clickhouse-url";
       private static final String CLICKHOUSE_USER = "clickhouse-user";
       private static final String CLICKHOUSE_PASSWORD = "clickhouse-password";
   
       public static void main(String[] args) {
           Properties properties = new Properties();
           properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_BOOTSTRAP_SERVERS);
           properties.put(ConsumerConfig.GROUP_ID_CONFIG, "clickhouse-consumer-group");
           properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
           properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
   
           try (Consumer<String, String> consumer = new KafkaConsumer<>(properties)) {
               consumer.subscribe(Collections.singletonList(KAFKA_TOPIC));
               while (true) {
                   ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                   records.forEach(record -> processKafkaMessage(record.value()));
               }
           } catch (Exception e) {
               e.printStackTrace();
           }
       }
   
       private static void processKafkaMessage(String message) {
           writeToClickHouse(message);
       }
   
       private static void writeToClickHouse(String message) {
           // 实现将数据写入 ClickHouse 的逻辑
       }
   }
   

优势：

• 实时性高
• 消息持久化

缺点：

• 一致性保证
• 对于小规模应用可能过于笨重