PostgreSQL传感器数据转储至Timescale时序表的高效方案问询

1. 替换Cron：用Node.js实现精确高频调度 #

直接用Node.js的递归setTimeout替代Cron，避免任务叠加问题（如果单次任务耗时超过500ms，setInterval会导致任务堆积）：

let isTaskRunning = false;

const syncData = async () => {
  if (isTaskRunning) return;
  isTaskRunning = true;
  
  try {
    // 执行数据同步逻辑
    await syncLiveToTimescale();
  } catch (err) {
    console.error('同步失败:', err);
  } finally {
    isTaskRunning = false;
    setTimeout(syncData, 500); // 上一次任务完成后，再间隔500ms启动下一次
  }
};

// 启动调度
syncData();

2. 数据库操作优化（适配100+传感器、50+标签规模） #

（1）增量读取data_live #

不要每次全表扫描，记录上次同步的时间戳，只拉取更新过的数据：

-- 假设data_live有updated_at字段记录数据更新时间
SELECT sensor_id, tag, value 
FROM data_live 
WHERE updated_at > $1; -- $1传入上次同步的时间戳

（2）预生成Crosstab查询模板 #

因为标签数量固定（50+），提前生成固定结构的crosstab语句，避免动态拼接SQL的开销：

-- 提前获取所有标签并生成模板，比如标签为temp、humidity、pressure等
SELECT * FROM crosstab(
  'SELECT sensor_id, tag, value FROM temp_sync_data',
  'SELECT DISTINCT tag FROM data_live ORDER BY tag'
) AS ct(
  sensor_id INT,
  temp FLOAT,
  humidity FLOAT,
  pressure FLOAT,
  -- 依次列出所有50+标签字段
);

可以在Node.js启动时预查询一次所有标签，动态生成这个SQL模板，后续直接复用。

（3）批量插入到Timescale #

用批量INSERT或PostgreSQL的COPY命令提升插入效率，避免单条插入的性能损耗：

INSERT INTO data_ts (time, sensor_id, temp, humidity, pressure)
VALUES 
  ($1, $2, $3, $4, $5),
  ($6, $7, $8, $9, $10),
  -- 批量插入多条数据
ON CONFLICT DO NOTHING; -- 避免重复插入相同时序数据

如果数据量极大，优先用COPY命令直接从内存写入，比INSERT效率高一个数量级。

（4）异步触发器替代定时查询（可选） #

如果data_live是传感器数据更新时实时写入/更新的，用PostgreSQL触发器+pg_notify实现实时同步，彻底替代定时拉取：

1. 在data_live上创建触发器，数据更新时发送通知：

CREATE OR REPLACE FUNCTION notify_data_update()
RETURNS TRIGGER AS $$
BEGIN
  PERFORM pg_notify('sensor_data_update', row_to_json(NEW)::TEXT);
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER trigger_data_update
AFTER UPDATE ON data_live
FOR EACH ROW EXECUTE FUNCTION notify_data_update();

2. 在Node.js中监听该频道，收到通知后直接处理行转列并插入data_ts，响应速度更快。

3. 大规模场景额外优化 #

• 缓存标签列表：启动时查询一次所有标签并缓存，后续无需每次执行DISTINCT tag查询。
• 数据库索引优化：给data_live的sensor_id、tag、updated_at创建联合索引；给Timescale的data_ts表按time和sensor_id创建分区，提升插入和查询性能。
• 分批次处理：如果单次同步数据量过大，拆分为多个小批次处理，避免内存占用过高和数据库锁等待。

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