如何快速同步同一服务器上两个PostgreSQL数据库的新增表数据？

嘿，你的这个场景我太有共鸣了——全量拉取所有ID到Python里做对比，数据量小的时候还好，一旦表行数上去，不仅占内存，还慢得让人头疼。既然两个数据库在同一台服务器上，咱们完全可以把这个对比工作交给PostgreSQL自己来做，效率能提升一大截！

方案一：跨数据库直接查询差异行 #

因为同服务器，PostgreSQL允许直接跨库访问表（前提是你的数据库用户有足够权限）。我们可以直接写一条SQL，从db1里找出所有ID不在db2的行，不用把所有数据拉到Python里处理。

优化后的代码示例 #

import psycopg2
from psycopg2.extras import execute_values

# 只需要连接其中一个库（比如db1），就能跨库访问db2
conn = psycopg2.connect(host="your_host", dbname="db1", user="your_user", password="your_pwd")
cursor = conn.cursor()

# 核心：直接在数据库层面找出需要同步的行
# 注意替换成你的实际schema（默认是public）和表名
sync_query = """
SELECT * FROM public.links l
WHERE NOT EXISTS (
    SELECT 1 FROM db2.public.links l2
    WHERE l2.id = l.id
)
"""

cursor.execute(sync_query)
new_rows = cursor.fetchall()

if new_rows:
    # 获取表的列名，自动生成插入语句（避免硬编码列名）
    cursor.execute("""
        SELECT column_name 
        FROM information_schema.columns 
        WHERE table_name = 'links' 
        ORDER BY ordinal_position
    """)
    columns = [col[0] for col in cursor.fetchall()]
    columns_str = ", ".join(columns)
    
    # 用execute_values做批量插入，比循环单条插入快N倍
    insert_query = f"""
        INSERT INTO db2.public.links ({columns_str})
        VALUES %s
    """
    execute_values(cursor, insert_query, new_rows)
    conn.commit()

cursor.close()
conn.close()

为什么这更快？ #

• 数据库本身对集合对比做了深度优化，尤其是如果id列有索引的话，NOT EXISTS的查询会非常高效。
• 不需要把所有ID拉到Python内存里做对比，减少了数据传输和内存占用。
• 批量插入(execute_values)避免了循环单条插入的网络开销和数据库事务开销。

方案二：基于时间/自增ID的增量同步（更优！） #

如果你的links表有创建时间字段（比如created_at）或者自增ID，那这个方案是效率最高的——我们只需要同步上次同步之后新增的行，完全不需要对比所有ID。

代码示例（以创建时间为例） #

import psycopg2
from psycopg2.extras import execute_values
import datetime

# 假设你把上次同步的时间存在一个文件里，或者单独的配置表中
def load_last_sync_time():
    try:
        with open("last_sync.txt", "r") as f:
            return datetime.datetime.fromisoformat(f.read().strip())
    except FileNotFoundError:
        # 第一次同步，取一个很早的时间
        return datetime.datetime(2000, 1, 1)

def save_last_sync_time(sync_time):
    with open("last_sync.txt", "w") as f:
        f.write(sync_time.isoformat())

# 主逻辑
last_sync_time = load_last_sync_time()
conn = psycopg2.connect(host="your_host", dbname="db1", user="your_user", password="your_pwd")
cursor = conn.cursor()

# 只查上次同步之后新增的行
sync_query = """
SELECT * FROM public.links
WHERE created_at > %s
ORDER BY created_at
"""

cursor.execute(sync_query, (last_sync_time,))
new_rows = cursor.fetchall()

if new_rows:
    # 获取列名，批量插入到db2
    cursor.execute("""
        SELECT column_name 
        FROM information_schema.columns 
        WHERE table_name = 'links' 
        ORDER BY ordinal_position
    """)
    columns = [col[0] for col in cursor.fetchall()]
    columns_str = ", ".join(columns)
    
    insert_query = f"""
        INSERT INTO db2.public.links ({columns_str})
        VALUES %s
    """
    execute_values(cursor, insert_query, new_rows)
    conn.commit()
    
    # 更新上次同步时间为本次同步的最大created_at
    cursor.execute("SELECT MAX(created_at) FROM public.links WHERE created_at > %s", (last_sync_time,))
    latest_time = cursor.fetchone()[0]
    save_last_sync_time(latest_time)

cursor.close()
conn.close()

这个方案的优势 #

• 数据量越大，优势越明显——每次只处理新增的少量行，查询和插入都极快。
• 避免了全表扫描和跨表对比的开销，几乎是最优的同步方式（只要有合适的增量标识字段）。

方案三：PostgreSQL内置工具（适合无自定义逻辑的场景） #

如果你的同步逻辑很简单（就是全量同步新增行，不需要额外处理），可以直接用PostgreSQL的内置工具：

• 逻辑复制：PostgreSQL 10+支持逻辑复制，可以把db1的links表实时同步到db2，不需要自己写脚本，配置好之后自动运行。
• pg_dump + 定时任务：用pg_dump导出db1的新增数据，再导入到db2，适合定时同步的场景，但灵活性不如自定义脚本。

注意事项 #

1. 权限问题：确保你的数据库用户有跨库访问的权限，比如db1的用户要能读取db2的links表，或者有插入db2的权限。
2. 索引优化：如果id或者created_at列没有索引，一定要加上，不然查询会很慢。
3. 分批处理：如果单次同步的行特别多，不要用fetchall()，改用fetchmany(size=1000)分批处理，避免内存溢出。

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