批量处理

从逐条处理到批量处理

还记得我最初的实现吗？

每次取消订单都要：

1. 更新订单状态（数据库写入）
2. 释放库存（数据库写入）
3. 发送通知（HTTP 请求）
4. 记录日志

如果有 1000 个订单要取消，那就是 3000 次数据库写入和 1000 次 HTTP 请求！

逐条处理的性能瓶颈：

数据库 I/O：3000 次
网络 I/O：1000 次
总耗时：~15 秒

---

批量处理的优势

性能对比

原理

逐条处理：
┌─────────┐    ┌─────────┐    ┌─────────┐
│ 任务 1  │───►│ 处理任务  │───►│ 写数据库  │
└─────────┘    └─────────┘    └─────────┘
┌─────────┐    ┌─────────┐    ┌─────────┐
│ 任务 2  │───►│ 处理任务  │───►│ 写数据库  │
└─────────┘    └─────────┘    └─────────┘
...

批量处理：
┌─────────┐    ┌─────────┐    ┌─────────┐
│ 任务 1  │────│          │    │         │
├─────────┤    │ 批量处理  │───►│ 批量写入  │
│ 任务 2  │────│          │    │         │
├─────────┤    └─────────┘    └─────────┘
│ 任务 3  │────│
└─────────┘    │
...           │
              │
              ▼
         一次处理完成

---

批量处理实现

数据库批量更新

取消订单这类操作最终都要落到 UPDATE 上。逐条更新的问题是往返次数：100 个订单就是 100 次网络往返，即使每条 SQL 在服务器上只花 0.5ms，光往返就能吃掉 50ms。

方案 A：IN + 统一值（最常用）

一批任务状态相同时，没必要拼 CASE WHEN，一条 IN 就够了：

UPDATE delay_tasks
SET status = 'DONE',
    updated_at = NOW()
WHERE id IN (?, ?, ?, ?)
  AND status = 'PENDING';

方案 B：IN + CASE WHEN（每行取值不同）

订单取消的原因五花八门，每行要写的值不一样时，用 CASE WHEN 把多行合并成一条 SQL：

UPDATE delay_tasks
SET status        = 'CANCELLED',
    updated_at    = NOW(),
    error_message = CASE
        WHEN id = 10001 THEN '用户主动取消'
        WHEN id = 10002 THEN '超时未支付'
        WHEN id = 10003 THEN '风控拦截'
        ELSE error_message
    END
WHERE id IN (10001, 10002, 10003)
  AND status = 'PENDING';

这里有两个容易踩的坑。第一个是 WHERE 里必须带上状态条件：少了 status = 'PENDING'，批量更新会覆盖掉别的进程刚写进去的中间状态，任务就会重复执行——这正是《技术挑战》里”挑战 2：任务重复执行”那类事故的来源。第二个是 SQL 长度：CASE WHEN 的文本随行数线性增长，MySQL 单条 SQL 受 max_allowed_packet 限制（默认 4MB），行数一多就必须拆批，或者干脆退回方案 A。

用 JDBC 的 Batch 提交更省事，一次 executeBatch() 把所有语句攒成一次网络往返：

try (Connection conn = dataSource.getConnection()) {
    conn.setAutoCommit(false);
    try (PreparedStatement ps = conn.prepareStatement(
            "UPDATE delay_tasks SET status = ?, updated_at = NOW() "
          + "WHERE id = ? AND status = 'PENDING'")) {
        for (DelayTask task : batch) {
            ps.setString(1, "DONE");
            ps.setLong(2, task.id());
            ps.addBatch();
        }
        ps.executeBatch();   // 一次往返
    }
    conn.commit();
}

---

批量查询

批量查询要消灭的是 N+1 查询：调度线程每轮捞出 100 个任务，如果拿着这 100 个 id 逐个回查订单详情，就是 1 + 100 次数据库往返。

用 IN 代替循环查询：

SELECT id, order_no, user_id, amount, status
FROM orders
WHERE id IN (?, ?, ?, ?);

能 JOIN 就不要”查出来再拼”：

SELECT t.id       AS task_id,
       t.task_type,
       o.order_no,
       o.user_id,
       o.amount
FROM delay_tasks t
JOIN orders o ON o.id = t.order_id
WHERE t.status = 'PENDING'
  AND t.execute_at <= NOW()
LIMIT 200;

更关键的一步是把”查到”和”占住”合成一条语句。先查出来、回头再 UPDATE，中间存在一个窗口期，别的消费者很可能把同一批任务抢走。MySQL 8.0 提供了 FOR UPDATE SKIP LOCKED，可以直接跳过已被别的事务锁住的行：

START TRANSACTION;

SELECT id, task_type, task_data
FROM delay_tasks
WHERE status = 'PENDING'
  AND execute_at <= NOW()
ORDER BY execute_at
LIMIT 200
FOR UPDATE SKIP LOCKED;

-- 用上一步拿到的 id 批量更新
UPDATE delay_tasks
SET status = 'PROCESSING', consumer_id = ?
WHERE id IN (?, ?, ?);

COMMIT;

SKIP LOCKED 是这套方案能横向扩展的关键：多个调度线程同时跑，各自拿到互不重叠的一批任务，不必互相等待。代价是它要求 MySQL 8.0 以上，5.7 只能靠应用层的分布式锁绕开，具体做法见《锁机制》。

---

批量插入

批量插入的收益和批量更新一样直接：1000 条任务从 1000 次往返降到 1 次。

多值 INSERT（最快）：

INSERT INTO delay_tasks
  (task_id, task_type, task_data, execute_at, status)
VALUES
  ('T1001', 'ORDER_CANCEL', '{"orderId":1}', NOW() + INTERVAL 30 MINUTE, 'PENDING'),
  ('T1002', 'ORDER_CANCEL', '{"orderId":2}', NOW() + INTERVAL 30 MINUTE, 'PENDING'),
  ('T1003', 'AUTO_CONFIRM', '{"orderId":3}', NOW() + INTERVAL 7 DAY,     'PENDING');

幂等插入：

延时任务经常被重复提交（用户重复点击、客户端自动重试），在 task_id 上建唯一索引，用 ON DUPLICATE KEY UPDATE 让重复提交变成一次无害的更新：

INSERT INTO delay_tasks
  (task_id, task_type, task_data, execute_at, status)
VALUES (?, ?, ?, ?, 'PENDING')
ON DUPLICATE KEY UPDATE
  execute_at   = VALUES(execute_at),
  task_data    = VALUES(task_data),
  status       = 'PENDING',
  retry_count  = 0;

这里有个容易忽略的细节：ON DUPLICATE KEY UPDATE 的冲突判定只认唯一索引，不认业务字段。所以 task_id 上一定要建唯一索引，否则”幂等”只是错觉。另外 MySQL 8.0.20 之后 VALUES(col) 这种写法已经废弃，推荐换成 AS new 别名写法。

---

批量处理的注意事项

1. 批量大小的选择

批量大小建议：

2. 事务管理

批量操作最容易出事的地方就是事务：包得太大，锁得久，并发就崩；包得太小，一半成功一半失败，数据就花了。

先明确 autocommit 的默认值。 MySQL 默认 autocommit = ON，每条语句都是独立事务，写完立刻提交。对逐条处理来说这是对的；对批量处理来说，这意味着 100 次更新就是 100 次提交、100 次刷盘，批量带来的收益被提交开销吃掉了大半。所以批量场景一定要显式关掉自动提交：

SET autocommit = 0;
-- 一批 UPDATE ...
COMMIT;   -- 或 ROLLBACK

事务边界应该划在哪里？ 原则是”一次业务语义 = 一个事务”，而不是”一批 = 一个事务”。同一批里的 100 个订单取消，如果它们之间没有一致性依赖，就应该拆成若干个事务；只有当它们必须”要么全成功要么全失败”时才合成一个大事务。

提交频率与锁持有时间的取舍。 事务开着不提交，InnoDB 的行锁和间隙锁就一直持有，其他事务只能干等。延时任务表的热点行往往集中在”即将到期”的那一小批记录上，锁竞争会直接表现为调度线程的 P99 延迟飙升。经验值是：

• 批量大小控制在 100~500 行；
• 事务持有时间控制在 100ms 以内；
• 超过这个量就拆成多个事务逐个提交。

回滚行为。 ROLLBACK 会撤销整个事务里所有的修改，所以它只适用于”整批都不能接受”的情况。真实业务里更常见的是部分失败：100 个订单里 3 个失败了，剩下 97 个不该跟着回滚。这时有两条路：

-- 路线 1：先落库，再重试失败项（推荐）
-- 97 条成功，3 条失败的单独记入重试表 / 死信队列
UPDATE delay_tasks SET status = 'DONE' WHERE id IN (97 个 id);
INSERT INTO delay_task_failed (task_id, error_message, retry_count)
VALUES (?, ?, 0), (?, ?, 0), (?, ?, 0);

-- 路线 2：全部回滚，整批重做
-- 代价是那 97 个已经成功的业务副作用（发过的通知、扣过的库存）无法回滚

关键点在于：数据库事务只能回滚数据库的修改，回滚不了已经发出去的副作用。所以凡是涉及外部调用的批量操作，正确姿势都是”数据库先提交、外部调用放在事务外做、失败了再补偿”，而不是”把外部调用塞进事务里，指望回滚兜底”。

最后是隔离级别。 默认的 REPEATABLE READ 在 SELECT ... FOR UPDATE 时可能因为间隙锁造成幻读，批量场景建议显式使用 READ COMMITTED（间隙锁更少、死锁更少），并把 innodb_lock_wait_timeout 设成秒级而不是默认的 50 秒，避免一个卡住的事务把整批调度拖死。

---

3. 错误处理

批量处理最难的不是把它写快，而是处理”部分成功”这个状态。100 条 SQL 里第 37 条超时了，剩下的 99 条算成功还是失败？重试第 37 条会不会把第 12 条重复执行一遍？

第一步：把错误分级。

第二步：让单条失败不拖垮整批。 executeBatch() 抛异常时，JDBC 并不会告诉你具体是哪一条失败了，可靠做法是自己检查 BatchUpdateException 里的 updateCounts：

try {
    ps.executeBatch();
    conn.commit();
} catch (BatchUpdateException e) {
    int[] counts = e.getUpdateCounts();
    List<Long> failed = new ArrayList<>();
    for (int i = 0; i < counts.length; i++) {
        if (counts[i] == Statement.EXECUTE_FAILED) {
            failed.add(ids.get(i));
        }
    }
    conn.rollback();
    // 成功的部分重新提交，失败的部分进重试队列
    ...
}

在 MySQL 上更简单的做法是不用大事务：每条语句独立提交，失败的单独重试。代价是失去批内原子性——但对延时任务来说，批内原子性本来就不是必需的，每条任务的状态是独立的。

第三步：重试要有退避和上限。 数据库死锁是高频事件，无脑立即重试只会让死锁更频繁。指数退避加随机抖动是标准做法：

long backoff = baseDelayMs * (1L << attempt) + ThreadLocalRandom.current().nextLong(100);

超过 max_retries 之后，任务不能就这么丢掉——把它挪进死信队列（dead letter queue），让人能查、能人工重放。“失败了但没人知道”是比”失败了”严重得多的问题。

第四步：错误要能定位到任务粒度。 批量场景下光记一条 “batch failed” 是不够的，必须记清楚：哪个 task_id、哪条 SQL、什么错误码、已经重试了几次。没有这些，凌晨三点排查就只能靠猜。

---

批量处理的完整示例

把前面几节拼起来，一个完整的批量调度器长这样：

调度线程（单线程，或加锁后的多线程）
  │
  ├─ 1. 抢任务：一条 SELECT ... FOR UPDATE SKIP LOCKED 拿 200 条
  │      └─ 拿到 task_id 列表 + 业务数据
  │
  ├─ 2. 批量占用：一条 IN + CASE WHEN 把状态改成 PROCESSING
  │      └─ 同一事务内提交，锁在这一刻就释放
  │
  ├─ 3. 逐条执行业务逻辑（释放库存、发通知）
  │      └─ 放在事务外，失败不影响已提交的状态
  │
  ├─ 4. 批量回写：一条 IN + CASE WHEN 写 DONE / FAILED
  │      └─ 失败的进重试表，超过上限进死信队列
  │
  └─ 5. 记录指标：本批耗时、成功数、失败数、P99

对应的代码骨架：

public int processBatch(List<DelayTask> tasks) {
    if (tasks.isEmpty()) {
        return 0;
    }
    int ok = 0;
    try (Connection conn = dataSource.getConnection()) {
        conn.setAutoCommit(false);

        // 1. 批量占用（抢占 + 占用在同一个事务里）
        int claimed = claimTasks(conn, tasks);
        if (claimed == 0) {
            conn.rollback();
            return 0;
        }

        // 2. 批量执行业务 —— 刻意放在事务外，避免长时间持锁
        Map<Long, String> results = new LinkedHashMap<>();
        for (DelayTask task : tasks) {
            try {
                handler(task.type()).handle(task);        // 发通知、释放库存
                results.put(task.id(), "DONE");
                ok++;
            } catch (BizException e) {
                results.put(task.id(), "SKIPPED");        // 业务性失败不算错误
            } catch (Exception e) {
                results.put(task.id(), "FAILED");
                retryQueue.offer(task, e);               // 退避重试，超限进死信
            }
        }

        // 3. 批量回写
        writeBack(conn, results);

        conn.commit();
    } catch (SQLException e) {
        log.error("批次处理失败, size={}", tasks.size(), e);
        // 整批回滚后重新入队，下一轮再试
    }
    return ok;
}

private void writeBack(Connection conn, Map<Long, String> results) throws SQLException {
    StringBuilder sql = new StringBuilder(
            "UPDATE delay_tasks SET status = CASE id ");
    for (Long id : results.keySet()) {
        sql.append("WHEN ").append(id).append(" THEN ? ");
    }
    sql.append("ELSE status END, updated_at = NOW() WHERE id IN (");
    sql.append("?,".repeat(results.size() - 1)).append("?)");

    try (PreparedStatement ps = conn.prepareStatement(sql.toString())) {
        int i = 1;
        for (String v : results.values()) {
            ps.setString(i++, v);
        }
        for (Long id : results.keySet()) {
            ps.setLong(i++, id);
        }
        ps.executeUpdate();
    }
}

要特别注意的是第 2 步和第 3 步的位置：业务逻辑在事务外。把外部调用放进事务里，一是长时间持锁，二是外部调用根本不受数据库事务管辖，回滚也回滚不掉。

效果对一下账：1000 个任务，数据库侧从 3000 次往返降到 3 次（抢占 1 次、占用 1 次、回写 1 次），业务侧的 1000 次外部调用无法被批量化（除非对方也提供批量接口）。总耗时从「批量处理的优势」一节估算的 15 秒降到 2 秒左右——收益几乎全部来自数据库往返的减少，这也是为什么批量优化的第一优先级永远是”减少往返次数”。

---

想一想

思考 1

批量处理时，如果部分任务失败，应该如何处理？

上一章节

索引优化

下一章节

Redis ZSet 方案