如何检测PostgreSQL远程库数据变更并同步至本地MySQL库？

嘿，我来帮你搞定远程PostgreSQL到本地MySQL的users表同步问题！结合你给出的代码片段，我整理了几个实用的方案，从轻量到专业场景都覆盖到了：

这是最适合中小场景的方案，核心思路是靠变更时间标记只同步远程库中新增/修改的记录，再处理删除逻辑。

前置准备 #

首先得确保远程PostgreSQL的users表有一个自动更新的时间戳字段，比如updated_at：

-- 给PostgreSQL的users表加字段+触发器实现自动更新
ALTER TABLE users ADD COLUMN updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP;

CREATE OR REPLACE FUNCTION update_updated_at()
RETURNS TRIGGER AS $$
BEGIN
    NEW.updated_at = CURRENT_TIMESTAMP;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER trigger_update_users_updated_at
BEFORE UPDATE ON users
FOR EACH ROW EXECUTE FUNCTION update_updated_at();

优化你的Go代码逻辑 #

你可以把原来的全量对比改成增量拉取+映射对比，效率会高很多：

func SyncUsers() {
    // 先从本地存储（比如配置文件/本地数据库）获取上次同步的时间
    lastSyncTime := getLastSyncTime()
    
    // 只拉取PostgreSQL中上次同步后变更的用户（新增/修改）
    changedPostgresUsers, err := GetUsersPostgresSince(lastSyncTime)
    if err != nil {
        // 这里记得加错误处理
        log.Printf("拉取PostgreSQL变更用户失败: %v", err)
        return
    }

    // 把本地MySQL的用户转成map，方便快速查找
    mysqlUsers, err := GetUsersMysql()
    if err != nil {
        log.Printf("拉取MySQL用户失败: %v", err)
        return
    }
    mysqlUserMap := make(map[int]User) // 假设主键是int类型的user_id
    for _, mu := range mysqlUsers {
        mysqlUserMap[mu.ID] = mu
    }

    // 处理新增和更新
    for _, pu := range changedPostgresUsers {
        if mu, exists := mysqlUserMap[pu.ID]; exists {
            // 对比核心字段，有变化才更新（避免无意义的写操作）
            if mu.Name != pu.Name || mu.Email != pu.Email || mu.Phone != pu.Phone {
                if err := UpdateUserMysql(pu); err != nil {
                    log.Printf("更新用户ID:%d失败: %v", pu.ID, err)
                }
            }
        } else {
            // 本地没有，插入新用户
            if err := InsertUserMysql(pu); err != nil {
                log.Printf("插入用户ID:%d失败: %v", pu.ID, err)
            }
        }
        // 移除已处理的用户，剩下的就是本地可能需要删除的
        delete(mysqlUserMap, pu.ID)
    }

    // 处理删除：剩下的mysqlUserMap里的用户，是远程库已经删除的
    // 这里推荐用软删除（比如给MySQL加is_deleted字段），避免误删
    for _, mu := range mysqlUserMap {
        if err := SoftDeleteUserMysql(mu.ID); err != nil {
            log.Printf("软删除用户ID:%d失败: %v", mu.ID, err)
        }
        // 如果要物理删除：DeleteUserMysql(mu.ID)
    }

    // 更新上次同步时间为当前时间
    if err := updateLastSyncTime(time.Now()); err != nil {
        log.Printf("更新同步时间失败: %v", err)
    }
}

如果需要接近实时的同步，逻辑复制是更可靠的选择，它能捕获PostgreSQL的所有变更（包括删除），不需要依赖时间戳。

核心步骤 #

1. 修改PostgreSQL配置（需要重启数据库）：

   # postgresql.conf
   wal_level = logical
   max_replication_slots = 10 # 根据需要调整
   

2. 创建PostgreSQL发布者，指定users表：

   CREATE PUBLICATION users_publication FOR TABLE users;
   

3. 用Go写一个订阅程序：利用github.com/jackc/pgx这类库订阅PostgreSQL的变更事件，然后把INSERT/UPDATE/DELETE操作转换成对应的MySQL语句执行。

这种方案的优势是无遗漏捕获变更，实时性能做到秒级，适合对同步时效要求高的场景。

如果不想自己写代码处理各种边缘情况（比如并发变更、事务一致性），可以用现成的CDC（变更数据捕获）工具：

• Debezium：基于Kafka的CDC工具，能捕获PostgreSQL的变更，然后同步到MySQL，支持高并发和事务一致性；
• Flink CDC：实时流处理工具，配置好源（PostgreSQL）和目标（MySQL）就能自动同步，还能做数据清洗；
• DataX：阿里开源的离线同步工具，适合定时批量同步的场景。

这些工具已经封装了各种异常处理逻辑，稳定性和扩展性都比自己写代码强，适合生产环境使用。

额外注意点 #

• 主键一致性：一定要保证两个表的主键（比如user_id）完全一致，否则对比和同步会出错；
• 事务包裹：同步操作最好用事务包裹，比如批量插入/更新时，一旦出错可以回滚，避免数据不一致；
• 冲突处理：如果本地MySQL的users表也允许修改，要提前定义冲突规则（比如以远程PostgreSQL为准，或者触发人工审核）。

内容的提问来源于stack exchange，提问作者Medone