大流量事件表的去重与按entity_type分组查询优化咨询

场景背景 #

我维护着一个按event_time小时分区的events表，不同源表会向该表写入带entity_id和payload的事件数据，这些事件通过Java中继服务发布到Kafka Topic。每小时处理量超500万行，数据发布后标记processed=true，24小时后直接删除分区以规避单条删除的性能问题。

需求 #

• 为表添加去重逻辑：若存在相同entity_id和entity_type的未处理行，更新其payload而非新增事件
• 当前查询逻辑为where processed=false order by event_time asc limit 1000，需实现按entity_type分组查询
• 寻求大流量场景下，插入去重及按entity_type分组查询的高效方案

表结构 #

CREATE TABLE events ( 
  id BIGSERIAL NOT NULL, 
  entity_type TEXT NOT NULL, -- 源表标识
  entity_id TEXT NOT NULL, -- 源表主键
  event_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, 
  payload JSON, 
  processed BOOLEAN DEFAULT false, 
  PRIMARY KEY (id, event_time) 
) PARTITION BY RANGE (event_time);

解决方案 #

1. 插入去重逻辑（更新而非新增） #

针对未处理数据的去重，推荐两种适配高并发的方案：

方案A：原子化冲突处理（单条/小批量写入）

利用PostgreSQL的INSERT ... ON CONFLICT语法，配合部分唯一索引实现原子更新：

-- 创建仅约束未处理数据的唯一索引
CREATE UNIQUE INDEX idx_events_entity_unprocessed ON events (entity_type, entity_id) WHERE processed = false;

-- 插入/更新语句
INSERT INTO events (entity_type, entity_id, payload)
VALUES ('user', '123', '{"action": "update"}')
ON CONFLICT (entity_type, entity_id) WHERE processed = false
DO UPDATE SET payload = EXCLUDED.payload, event_time = CURRENT_TIMESTAMP;

该方案无需应用层额外逻辑，由数据库保证原子性，适合高并发单条写入场景。唯一索引会增加少量写入开销，但500万/小时的量级下，合理配置硬件可完全承受。

方案B：批量拆分处理（大批量写入）

如果是批量导入场景，可先查询已存在的未处理键，拆分数据为更新/插入两组分别操作，减少冲突锁竞争：

// 伪代码示例
List<Event> eventsToWrite = ...;
Set<EntityKey> existingKeys = queryExistingUnprocessedKeys(eventsToWrite);

List<Event> toInsert = eventsToWrite.stream()
    .filter(e -> !existingKeys.contains(new EntityKey(e.getEntityType(), e.getEntityId())))
    .collect(Collectors.toList());

List<Event> toUpdate = eventsToWrite.stream()
    .filter(e -> existingKeys.contains(new EntityKey(e.getEntityType(), e.getEntityId())))
    .collect(Collectors.toList());

// 批量执行插入和更新
batchInsert(toInsert);
batchUpdate(toUpdate);

需注意在事务中执行查询+写入，避免中间数据变更导致的冲突。

2. 按entity_type分组查询未处理数据 #

根据业务需求选择两种实现方式：

方式A：按类型分组取固定条数

若需要每个entity_type获取指定数量的未处理数据，用窗口函数实现：

SELECT id, entity_type, entity_id, event_time, payload
FROM (
    SELECT *,
           ROW_NUMBER() OVER (PARTITION BY entity_type ORDER BY event_time ASC) AS rn
    FROM events
    WHERE processed = false
) t
WHERE rn <= 100 -- 每个类型取100条，可按需调整
LIMIT 1000; -- 全局总条数限制

优化提示：需配合(processed, entity_type, event_time)复合索引，避免全表扫描。

方式B：按类型单独拉取

若中继服务可按类型独立处理，直接过滤查询更高效：

SELECT id, entity_type, entity_id, event_time, payload
FROM events
WHERE processed = false AND entity_type = 'user'
ORDER BY event_time ASC
LIMIT 1000;

该查询可直接命中(processed, entity_type, event_time)索引，性能最优。

3. 大流量场景下的整体优化 #

索引优化

• 去重索引：idx_events_entity_unprocessed (entity_type, entity_id) WHERE processed = false
• 查询索引：idx_events_processed_type_time (processed, entity_type, event_time)，同时优化过滤、分组、排序逻辑
• 分区维护：用pgpartman自动管理小时级分区，定期清理24小时前的分区，保持表体积可控

写入优化

• 批量操作：优先使用批量INSERT/UPDATE，减少网络往返和事务开销
• 事务调优：Java应用中关闭自动提交，合并多个操作到单个事务
• 数据库参数调整：增大work_mem（用于排序/窗口函数）、max_wal_size（减少WAL刷写频率）、maintenance_work_mem（用于索引创建）

查询优化

• 键集分页：替代LIMIT/OFFSET，避免大偏移量导致的性能问题：

-- 首次查询
SELECT id, entity_type, entity_id, event_time, payload
FROM events
WHERE processed = false
ORDER BY event_time ASC, id ASC
LIMIT 1000;

-- 后续查询（用上一次结果的最后一条event_time和id作为条件）
SELECT id, entity_type, entity_id, event_time, payload
FROM events
WHERE processed = false
  AND (event_time > '2024-05-20 10:00:00' OR (event_time = '2024-05-20 10:00:00' AND id > 12345))
ORDER BY event_time ASC, id ASC
LIMIT 1000;

• 并行处理：中继服务启动多线程，每个线程处理一个entity_type的事件，提升处理吞吐量

其他优化

• 读写分离：用PostgreSQL流复制实现读写分离，查询走从库，缓解主库压力
• 缓存辅助：用Redis缓存最近的未处理(entity_type, entity_id)键，减少数据库查询次数（需注意更新/标记处理后及时清理缓存）

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