数据库数据同步常用的5种实施方案

#设置主服务器的唯一标识
server-id = 1
#启用二进制日志，记录主数据库上的所有更改
log_bin = /var/log/mysql/mysql-bin.log
#指定要复制的数据库
binlog_do_db = your_database_name

#replication_user和replication_password替换成自己的用户名和密码
#创建用于复制的用户
create user 'replication_user'@'%' identified by 'replication_password';
#复制授权
grant replication slave on *.* to 'replication_user'@'%';
#刷新权更改应用
flush privileges;

SHOW MASTER STATUS;
  
mysql> show master status;
+------------------+----------+--------------+------------------+-------------------+
| File             | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set |
+------------------+----------+--------------+------------------+-------------------+
| mysql-bin.000001 |     6470 | your_name    |                  |                   |
+------------------+----------+--------------+------------------+-------------------+
1 row in set (0.00 sec)

server-id = 2

#配置从数据库连接到主数据库
change master to
master_host = 'master_host',
master_user = 'replication_user',
master_password = 'replication_password',

#从SHOW MASTER STATUS;中获取的File值。
master_log_file = 'master_log_file',
#从SHOW MASTER STATUS;中获取的Position值
master_log_pos = master_log_pos;

START SLAVE;

SHOW SLAVE STATUS\G

# 创建触发器
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 ;

# 向tb_order插入数据
INSERT INTO tb_order VALUES (4, 4, '2024-01-15 12:00:00', 104, 4, 150.25, '待支付', '567 Elm St, County', 'Credit Card', 'DISCOUNT_15', '2024-01-15 12:00:00', '2024-01-15 12:00:00');
# 查询tb_order_his，确保数据同步成功
SELECT * FROM tb_order_his;

# insert into tb_target select * from tb_sourceINSERT 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_timeFROM tb_order;

# MySQL 连接配置
database.hostname=mysql-host
database.port=3306
database.user=mysql-user
database.password=mysql-password
# Debezium 配置
connector.class=io.debezium.connector.mysql.MySqlConnector
tasks.max=1
database.server.id=1
database.server.name=my-app-connector
database.whitelist=mydatabase

debezium-connector-mysql my-connector.properties

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

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) {
        // 解析 Kafka 消息，获取变更数据
        // 将数据写入 ClickHouse
        writeToClickHouse(message);
    }
    private static void writeToClickHouse(String message) {
        // 实现将数据写入 ClickHouse 的逻辑
    }
}