如何循环流式读取Postgres行数据并处理非唯一时间戳增量同步

方案对比 #

先针对你提到的两个方案做优劣势分析：

• 方案1（新增自增id字段）：可行，但有额外改表成本。如果你的表支持修改结构，新增一个BIGSERIAL类型的自增主键作为全局唯一有序游标，后续每次查询WHERE id > 上次最大id ORDER BY id LIMIT 50000，逻辑简单无重复，性能也有保证，适合可以改表的场景。
• 方案2（时间戳回退+去重）：不推荐。一方面如果同一秒内数据量超过单次查询上限会出现丢数据，另一方面业务层要维护去重逻辑，额外开销大，长期运行稳定性差。

更优方案（无需改表） #

你的场景是交易流水表，通常都是append only（仅新增不修改历史数据），不需要修改表结构就能实现精准分页拉取，不会丢数据也不会有重复：

核心逻辑：用内置隐藏字段做游标分页 #

PostgreSQL每张表自带隐藏唯一字段ctid，用于标识行的物理存储位置，对于append only的表ctid永久不变。结合时间戳可以组成完美的唯一游标，配合PostgreSQL原生支持的行值（Row Value）比较功能，分页逻辑非常简单：

1. 首次启动查询条件：

SELECT *, ctid FROM 你的表名 
WHERE ts >= NOW() - INTERVAL '1 hour' 
ORDER BY ts, ctid 
LIMIT 50000

2. 保存最后一行的last_ts和last_ctid作为游标
3. 后续每次增量查询条件：

SELECT *, ctid FROM 你的表名 
WHERE (ts, ctid) > (@last_ts, @last_ctid)
ORDER BY ts, ctid
LIMIT 50000

这个方案能完美解决时间戳不唯一的问题，不需要改表，也不会出现重复或者丢数据的情况，只要你给ts字段建了B树索引，查询性能完全能满足要求。

启动阶段优化：用流式查询减少往返 #

你启动阶段要拉150万行数据，不需要自己分页循环查询，.NET的PostgreSQL驱动Npgsql原生支持流式查询，一次性发起查询，驱动会自动分批从服务端拉取数据，不会把全量数据加载到内存，F#实现示例：

open Npgsql
open System

let connString = "你的Postgres连接字符串"
use conn = new NpgsqlConnection(connString)
conn.Open()

// 启动阶段流式拉取最近1小时数据
use cmd = new NpgsqlCommand("SELECT instrument, ts, price, quantity, direction FROM 你的表名 WHERE ts >= NOW() - INTERVAL '1 hour' ORDER BY ts, ctid", conn)
// 开启流式读取，不需要加载全量结果到内存
use reader = cmd.ExecuteReader()
while reader.Read() do
    let instrument = reader.GetString(0)
    let ts = reader.GetDateTime(1)
    let price = reader.GetDouble(2)
    let quantity = reader.GetDouble(3)
    let direction = reader.GetInt32(4)
    // 处理你的数据逻辑

这种方式比你自己循环分页发起几十次查询的效率高很多，也更省资源。

额外优化建议 #

• 给ts字段建升序索引：CREATE INDEX idx_你的表名_ts ON 你的表名 (ts ASC);，能大幅提升查询效率。
• 日常增量拉取的间隔可以动态调整，不需要固定每秒一次：如果上次返回了50k行（满负载），就立刻发起下一次查询，尽快追平最新数据，追平后再降到1秒一次即可。

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