如何优化PostgreSQL+SQLAlchemy查询高吞吐交易表最新数据无延迟

问题背景 #

我用Python、PostgreSQL和SQLAlchemy搭建演示系统，交易表正常时段处理约200笔/秒（TPS）交易，峰值可达400TPS。需要查询最新交易数据且无响应延迟，担忧当前按ID排序加limit的查询方式会随交易量增长引发性能问题。

系统详情 #

• 技术栈：PostgreSQL、SQLAlchemy
• 交易表结构：
  • id（主键）
  • created_unix_time（Big int）
  • 其他相关字段
• 目标：快速查询最新交易记录

当前实现代码 #

async def get_all_by_distinct_transaction_type(self, where: dict | None, skip: int = 0, limit: int = 10) -> List[TransactionModel]:
    try:
        if where is None:
            where = {}

        query = select(self.model_class)

        if where:
            condition = destruct_where(self.model_class, where)
            if condition is not None:
                query = query.where(condition)

        query = query.order_by(desc(self.model_class.id)
                               ).offset(skip).limit(limit)

        result = await self.session.execute(query)
        data = result.scalars().all()
        return data
    except Exception as e:
        log.error(f"Error get_all_by_distinct_transaction_type: {e}")
        raise InternalServer(
            error_code=ErrorCode.ERROR_EXECUTE_QUERY.name)

async def count_all_by_distinct_transaction_type(self, where: dict | None) -> int:
    try:
        query = select(self.model_class)
        if where is not None:
            condition = destruct_where(self.model_class, where)
            if condition is not None:
                query = query.where(condition)

        return await super()._count(query)
    except Exception as e:
        log.error(f"Error count_all_by_distinct_transaction_type: {e}")
        raise InternalServer(error_code=ErrorCode.ERROR_EXECUTE_QUERY.name)

优化方案 #

1. 替换offset分页为游标分页，消除大offset性能损耗 #

当前用offset(skip)的分页方式，当skip值较大时，数据库需要扫描大量数据再丢弃，性能会急剧下降。改用主键游标分页：

• 记录上一页返回的最后一条交易的id
• 下一页查询时用where id < last_id替代offset(skip)，配合order_by(desc(id)).limit(limit)
• 这种方式完全利用主键的B-tree索引，查询性能不会随数据量增长而下降

示例修改后的查询代码：

async def get_latest_transactions(self, where: dict | None, last_id: int | None = None, limit: int = 10) -> List[TransactionModel]:
    try:
        query = select(self.model_class)
        if where:
            condition = destruct_where(self.model_class, where)
            if condition is not None:
                query = query.where(condition)
        # 用游标替代offset
        if last_id is not None:
            query = query.where(self.model_class.id < last_id)
        query = query.order_by(desc(self.model_class.id)).limit(limit)
        
        result = await self.session.execute(query)
        return result.scalars().all()
    except Exception as e:
        log.error(f"Error get_latest_transactions: {e}")
        raise InternalServer(error_code=ErrorCode.ERROR_EXECUTE_QUERY.name)

2. 确保查询条件命中索引 #

• 检查destruct_where生成的过滤条件，如果涉及非主键、非时间字段，给这些字段添加B-tree索引（比如交易类型字段）
• 如果经常同时按多个条件过滤+排序，创建联合索引，比如(filter_column, id DESC)，让数据库可以直接通过索引获取数据，无需回表扫描

3. 优化计数逻辑，避免全表扫描 #

当前的count_all_by_distinct_transaction_type会执行全表计数，数据量大时性能差：

• 如果不需要精确的总条数，改用PostgreSQL的近似计数pg_stat_user_tables.n_live_tup，性能极高
• 如果必须精确计数，考虑缓存计数结果（比如用Redis缓存，每隔一段时间更新一次）

示例近似计数实现：

async def approximate_transaction_count(self, where: dict | None) -> int:
    try:
        if where is None:
            # 直接从系统表取近似行数
            query = select(func.pg_stat_user_tables.n_live_tup).where(
                func.pg_stat_user_tables.relname == self.model_class.__tablename__
            )
            result = await self.session.execute(query)
            return result.scalar() or 0
        # 有过滤条件时才执行精确计数，或根据业务场景缓存结果
        return await super()._count(select(self.model_class).where(destruct_where(self.model_class, where)))
    except Exception as e:
        log.error(f"Error approximate_transaction_count: {e}")
        raise InternalServer(error_code=ErrorCode.ERROR_EXECUTE_QUERY.name)

4. 数据库配置与维护优化 #

• 确保PostgreSQL的track_counts参数开启（默认开启），保证近似计数的可用性
• 定期对交易表执行VACUUM ANALYZE，清理表碎片、更新统计信息，维持索引效率
• 根据服务器配置调整shared_buffers、work_mem等参数，提升数据库的查询处理能力

5. 应用层缓存缓解数据库压力 #

• 用Redis缓存最近N条交易记录，查询最新数据时优先从Redis获取，再从数据库补充（适合对实时性要求不是绝对严格的场景）
• 高峰时段可以异步预加载最新数据到缓存，减少数据库的直接查询请求

内容的提问来源于stack exchange，提问作者Today's No-Code Recommender