ETest-Vue-FastAPI/批次聚合查询性能优化方案.md

批次聚合查询性能优化方案

性能分析

当前聚合查询的性能问题

查询特点:

• GROUP BY batch_id
• 多个 SUM(CASE...) 聚合函数
• LEFT JOIN sys_user
• WHERE order_id IS NULL

性能瓶颈:

1. 全表扫描: 如果没有索引，需要扫描所有工单
2. 多次聚合计算: 每个批次都要计算 7 个统计值
3. JOIN 操作: 需要关联用户表

数据量估算

工单数量	批次数量	查询时间（无索引）	查询时间（有索引）
1,000	100	~50ms	~10ms
10,000	1,000	~500ms	~50ms
100,000	10,000	~5s	~500ms
1,000,000	100,000	~50s	~5s

优化方案

方案 1: 添加数据库索引（推荐）⭐⭐⭐⭐⭐

优点:

• 简单有效
• 不改变数据结构
• 查询速度提升 10-100 倍

实现:

-- 1. 为 batch_id 添加索引（最重要）
CREATE INDEX idx_batch_id ON test_work_order(batch_id);

-- 2. 为 order_id 添加索引（WHERE 条件）
CREATE INDEX idx_order_id ON test_work_order(order_id);

-- 3. 为 test_step 添加索引（过滤条件）
CREATE INDEX idx_test_step ON test_work_order(test_step);

-- 4. 组合索引（最优）
CREATE INDEX idx_batch_order_step ON test_work_order(batch_id, order_id, test_step);

性能提升:

• 100,000 条工单：从 5s 降到 500ms
• 1,000,000 条工单：从 50s 降到 5s

方案 2: 创建批次表（适合超大数据量）⭐⭐⭐⭐

优点:

• 查询速度最快
• 支持更复杂的批次管理
• 适合数据量 > 100 万

缺点:

• 需要维护批次表
• 数据可能不同步

实现:

-- 创建批次表
CREATE TABLE test_work_order_batch (
    id INT PRIMARY KEY AUTO_INCREMENT,
    batch_id BIGINT NOT NULL UNIQUE,
    batch_name VARCHAR(100),
    creator INT,
    create_time DATETIME,
    work_order_count INT DEFAULT 0,
    pending_count INT DEFAULT 0,
    testing_count INT DEFAULT 0,
    review1_count INT DEFAULT 0,
    review2_count INT DEFAULT 0,
    review3_count INT DEFAULT 0,
    completed_count INT DEFAULT 0,
    batch_status INT DEFAULT 0,
    INDEX idx_batch_id (batch_id),
    INDEX idx_batch_status (batch_status),
    INDEX idx_create_time (create_time)
);

维护方式:

1. 创建工单时，更新批次表
2. 更新工单状态时，更新批次表
3. 定时任务同步数据

方案 3: 物化视图（MySQL 8.0+）⭐⭐⭐

优点:

• 自动维护
• 查询速度快

缺点:

• MySQL 不原生支持物化视图
• 需要使用触发器模拟

实现:

-- 创建汇总表
CREATE TABLE test_work_order_batch_summary (
    batch_id BIGINT PRIMARY KEY,
    batch_name VARCHAR(100),
    work_order_count INT,
    pending_count INT,
    testing_count INT,
    review1_count INT,
    review2_count INT,
    review3_count INT,
    completed_count INT,
    batch_status INT,
    last_update DATETIME,
    INDEX idx_batch_status (batch_status)
);

-- 创建触发器（INSERT）
DELIMITER $$
CREATE TRIGGER trg_workorder_insert
AFTER INSERT ON test_work_order
FOR EACH ROW
BEGIN
    INSERT INTO test_work_order_batch_summary (batch_id, batch_name, work_order_count, ...)
    VALUES (NEW.batch_id, NEW.batch_name, 1, ...)
    ON DUPLICATE KEY UPDATE
        work_order_count = work_order_count + 1,
        testing_count = testing_count + IF(NEW.test_step = 1, 1, 0),
        ...;
END$$
DELIMITER ;

-- 创建触发器（UPDATE）
-- 创建触发器（DELETE）

方案 4: Redis 缓存（适合高并发）⭐⭐⭐⭐

优点:

• 查询速度极快（< 10ms）
• 减轻数据库压力
• 适合高并发场景

缺点:

• 需要维护缓存
• 可能有数据延迟

实现:

import redis
import json

class BatchCacheService:
    def __init__(self):
        self.redis_client = redis.Redis(host='localhost', port=6379, db=0)
        self.cache_ttl = 300  # 5分钟过期
    
    async def get_batch_list(self, test_step=None):
        # 生成缓存 key
        cache_key = f"batch_list:step_{test_step}"
        
        # 尝试从缓存获取
        cached_data = self.redis_client.get(cache_key)
        if cached_data:
            return json.loads(cached_data)
        
        # 缓存未命中，查询数据库
        batch_list = await Test_work_orderDao.get_batch_list(...)
        
        # 存入缓存
        self.redis_client.setex(
            cache_key,
            self.cache_ttl,
            json.dumps(batch_list)
        )
        
        return batch_list
    
    def invalidate_batch_cache(self, batch_id):
        """工单状态变更时，清除相关缓存"""
        # 清除所有批次列表缓存
        for key in self.redis_client.scan_iter("batch_list:*"):
            self.redis_client.delete(key)

方案 5: 分页 + 索引（当前最佳方案）⭐⭐⭐⭐⭐

优点:

• 实现简单
• 性能足够好
• 不需要额外维护

实现:

-- 添加索引
CREATE INDEX idx_batch_order_step ON test_work_order(batch_id, order_id, test_step);

-- 分页查询（只查询当前页的批次）
SELECT 
    batch_id,
    batch_name,
    COUNT(*) as work_order_count,
    SUM(CASE WHEN test_step = 1 THEN 1 ELSE 0 END) as testing_count,
    ...
FROM test_work_order
WHERE order_id IS NULL
GROUP BY batch_id, batch_name
ORDER BY batch_id DESC
LIMIT 10 OFFSET 0;  -- 每页 10 条

性能:

• 100,000 条工单，1,000 个批次
• 查询第 1 页：~50ms
• 查询第 10 页：~100ms

推荐方案

小型系统（< 10 万条工单）

方案 1: 添加索引 + 分页

CREATE INDEX idx_batch_order_step ON test_work_order(batch_id, order_id, test_step);

中型系统（10 万 - 100 万条工单）

方案 1 + 方案 4: 索引 + Redis 缓存

# 1. 添加索引
# 2. 使用 Redis 缓存批次列表
# 3. 工单状态变更时清除缓存

大型系统（> 100 万条工单）

方案 2: 创建批次表

-- 1. 创建批次表
-- 2. 创建工单时更新批次表
-- 3. 更新工单状态时更新批次表
-- 4. 定时任务同步数据

立即优化（SQL 脚本）

-- 批次聚合查询性能优化
-- 执行时间: 2026-01-09

-- 1. 检查当前索引
SHOW INDEX FROM test_work_order;

-- 2. 添加批次ID索引（如果不存在）
CREATE INDEX IF NOT EXISTS idx_batch_id ON test_work_order(batch_id);

-- 3. 添加订单ID索引（如果不存在）
CREATE INDEX IF NOT EXISTS idx_order_id ON test_work_order(order_id);

-- 4. 添加测试步骤索引（如果不存在）
CREATE INDEX IF NOT EXISTS idx_test_step ON test_work_order(test_step);

-- 5. 添加组合索引（最优，如果不存在）
CREATE INDEX IF NOT EXISTS idx_batch_order_step 
ON test_work_order(batch_id, order_id, test_step);

-- 6. 分析表（更新统计信息）
ANALYZE TABLE test_work_order;

-- 7. 测试查询性能
EXPLAIN SELECT 
    batch_id,
    batch_name,
    COUNT(*) as work_order_count,
    SUM(CASE WHEN test_step = 1 THEN 1 ELSE 0 END) as testing_count
FROM test_work_order
WHERE order_id IS NULL
GROUP BY batch_id, batch_name
ORDER BY batch_id DESC
LIMIT 10;

性能对比

方案	10万工单	100万工单	实现难度	维护成本
无优化	5s	50s	-	-
索引	500ms	5s	低	低
索引+缓存	10ms	10ms	中	中
批次表	50ms	100ms	高	高
物化视图	100ms	200ms	高	中

结论

当前阶段推荐:

1. 立即添加索引（5 分钟实现，性能提升 10 倍）
2. 使用分页查询（已实现）
3. 监控查询性能

未来优化:

• 数据量 > 10 万时，考虑 Redis 缓存
• 数据量 > 100 万时，考虑批次表

完成时间

2026-01-09 01:15