如何用Python高效读取PostgreSQL实时新增行且无遗漏？

能否实现无遗漏读取？ #

可以实现无遗漏读取所有新增行，核心是要跟踪已读取的位置，避免每次只取最新一行而跳过中间数据。

原脚本的问题分析 #

原脚本的漏读并非cur.execute()执行耗时导致，而是逻辑设计错误：

• 每次仅执行SELECT * from postgrestable ORDER BY TIMESTAMP DESC LIMIT 1，只能获取当前最新的一行，两次查询之间插入的所有行都会被完全跳过。
• 存在语法错误：Python中布尔值应为True而非TRUE。
• 无休眠逻辑，会疯狂占用CPU资源，同时频繁查询给数据库带来不必要的压力。
• 未考虑事务隔离级别，默认READ COMMITTED下可能无法及时看到刚提交的新行。

可行的Python实现脚本 #

方案1：基于自增主键跟踪读取位置（通用方案） #

假设表有自增主键id（优先选择，比时间戳更可靠），通过记录最后读取的id，每次读取该位置之后的所有新增行：

import psycopg2
import time
from psycopg2.extras import RealDictCursor

# 替换为你的数据库连接参数
db_params = {
    "dbname": "your_database",
    "user": "your_user",
    "password": "your_password",
    "host": "127.0.0.1",
    "port": "5432"
}

def read_real_time_data():
    last_read_id = 0  # 初始值设为表中最小id以下，或根据历史数据调整
    conn = None
    try:
        print("\n连接PostgreSQL数据库...\n")
        conn = psycopg2.connect(**db_params)
        # 设置事务隔离级别，确保能读取已提交的新行
        conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_READ_COMMITTED)
        cur = conn.cursor(cursor_factory=RealDictCursor)

        while True:
            # 读取last_read_id之后的所有新增行，按id升序保证顺序
            cur.execute("SELECT * FROM postgrestable WHERE id > %s ORDER BY id ASC", (last_read_id,))
            rows = cur.fetchall()

            for row in rows:
                print(row)
                last_read_id = row["id"]  # 更新最后读取的位置

            # 根据写入频率调整休眠时间，平衡实时性与资源占用
            time.sleep(0.1)

    except Exception as e:
        print(f"错误：{str(e)}")
    finally:
        if conn:
            cur.close()
            conn.close()
            print("\n数据库连接已关闭。")

if __name__ == "__main__":
    read_real_time_data()

方案2：使用PostgreSQL LISTEN/NOTIFY机制（高效方案） #

如果写入端可以配合，在插入数据后发送NOTIFY通知，读取端监听通知后再查询新增行，减少空查询的资源消耗：

写入端触发通知（SQL示例）

在插入数据的事务中添加通知：

INSERT INTO postgrestable (col1, col2, timestamp) VALUES ('value1', 'value2', NOW());
NOTIFY new_data_arrived;

读取端Python脚本

import psycopg2
import select
from psycopg2.extras import RealDictCursor

db_params = {
    "dbname": "your_database",
    "user": "your_user",
    "password": "your_password",
    "host": "127.0.0.1",
    "port": "5432"
}

def listen_real_time_data():
    last_read_id = 0
    conn = None
    try:
        print("\n连接PostgreSQL数据库...\n")
        conn = psycopg2.connect(**db_params)
        # 监听NOTIFY必须设置自动提交
        conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
        cur = conn.cursor(cursor_factory=RealDictCursor)

        # 订阅通知频道
        cur.execute("LISTEN new_data_arrived;")
        print("已订阅new_data_arrived通知频道...")

        while True:
            # 等待通知，超时1秒防止永久阻塞
            if select.select([conn], [], [], 1) == ([], [], []):
                # 超时后主动查询一次，避免通知丢失
                cur.execute("SELECT * FROM postgrestable WHERE id > %s ORDER BY id ASC", (last_read_id,))
                rows = cur.fetchall()
                for row in rows:
                    print(row)
                    last_read_id = row["id"]
            else:
                # 处理收到的通知
                conn.poll()
                while conn.notifies:
                    notify = conn.notifies.pop()
                    print(f"收到通知：{notify.channel}")
                    # 查询新增行
                    cur.execute("SELECT * FROM postgrestable WHERE id > %s ORDER BY id ASC", (last_read_id,))
                    rows = cur.fetchall()
                    for row in rows:
                        print(row)
                        last_read_id = row["id"]

    except Exception as e:
        print(f"错误：{str(e)}")
    finally:
        if conn:
            cur.close()
            conn.close()
            print("\n数据库连接已关闭。")

if __name__ == "__main__":
    listen_real_time_data()

关键注意事项 #

• 优先使用自增主键id跟踪位置，避免时间戳重复导致的漏读/重复读取问题。
• 若脚本需要重启，建议将last_read_id保存到本地文件或缓存中，避免重启后重新读取全量数据。
• 根据实际写入频率调整休眠时间或通知触发逻辑，平衡实时性与系统资源占用。

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