Python与MySQL实现数据库实时同步的详细步骤

pip install pymysql mysql-connector-python pymysqlreplication

[mysqld]
log-bin=mysql-bin        # 开启 binlog 功能
server-id=1              # 唯一标识符，必须设置
binlog-format=row        # 使用行级日志，便于捕获数据变更

SHOW VARIABLES LIKE 'log_bin';
SHOW VARIABLES LIKE 'binlog_format';

from pymysqlreplication import BinLogStreamReader
from pymysqlreplication.row_event import DeleteRowsEvent, WriteRowsEvent, UpdateRowsEvent
import pymysql

# 源数据库配置
SOURCE_CONFIG = {
    "host": "localhost",
    "port": 3306,
    "user": "root",
    "passwd": "password"
}

# 目标数据库配置
TARGET_CONFIG = {
    "host": "localhost",
    "port": 3306,
    "user": "root",
    "passwd": "password",
    "database": "target_db"
}

# 连接目标数据库
def write_to_target_db(query, params):
    connection = pymysql.connect(**TARGET_CONFIG)
    try:
        with connection.cursor() as cursor:
            cursor.execute(query, params)
        connection.commit()
    finally:
        connection.close()

# 处理 binlog 事件
def process_binlog_event():
    stream = BinLogStreamReader(
        connection_settings=SOURCE_CONFIG,
        server_id=100,   # 唯一 server_id
        blocking=True,   # 持续监听
        only_events=[WriteRowsEvent, UpdateRowsEvent, DeleteRowsEvent]
    )

    for binlogevent in stream:
        for row in binlogevent.rows:
            if isinstance(binlogevent, WriteRowsEvent):
                # INSERT 事件
                query = "INSERT INTO target_table (id, name, age) VALUES (%s, %s, %s)"
                params = (row["values"]["id"], row["values"]["name"], row["values"]["age"])
                write_to_target_db(query, params)

            elif isinstance(binlogevent, UpdateRowsEvent):
                # UPDATE 事件
                query = "UPDATE target_table SET name=%s, age=%s WHERE id=%s"
                params = (row["after_values"]["name"], row["after_values"]["age"], row["after_values"]["id"])
                write_to_target_db(query, params)

            elif isinstance(binlogevent, DeleteRowsEvent):
                # DELETE 事件
                query = "DELETE FROM target_table WHERE id=%s"
                params = (row["values"]["id"],)
                write_to_target_db(query, params)

    stream.close()

# 启动数据同步
if __name__ == "__main__":
    process_binlog_event()