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()