本节目标：把「一条一条处理」改成可重入、可观测的批处理，并用真实数据量出差距。
适用版本：JDK 25（Temurin 25.0.5）；PostgreSQL 17.11（postgres:17-alpine）；PostgreSQL JDBC 42.7.14。
承接的问题
罚金在归还时按天计算，但在实际系统里，结算往往是一个独立任务：月末把上个月所有已归还、未结算的借阅记录找出来，按 due_on 和 returned_on 的差值算罚金，写回 loans.fine_cents。libraryd 的 FinePolicy.dailyFineCents() 是每天 50 分，计算本身很简单，难的是数据量。
假设有 6000 条借阅记录，其中 5400 条已归还。最直接的写法是查出来，循环每一条执行一个 update：
try (PreparedStatement ps = connection.prepareStatement(
"update loans set fine_cents = ? where id = ?")) {
for (LoanFineRow row : rows) {
ps.setLong(1, fineCents(row));
ps.setLong(2, row.id());
ps.executeUpdate();
}
}
代码没有错，但在本地 PostgreSQL 上，6000 条插入要十几秒，5400 条更新也要十几秒。原因不是 SQL 慢，而是每条 executeUpdate 都在自动提交模式下走一次事务提交，PostgreSQL 要为每次提交做 WAL 刷盘。把 500 条合成一批、放在一个事务里，时间会降到几百毫秒。
这不是「JDBC 批处理 API 更快」的魔法，而是「减少提交次数」的收益。理解这一点，才能判断什么时候该分批、批多大、失败了怎么重跑。
逐条与分批的真实差距
先准备一个可复现的实验。用独立容器跑 PostgreSQL 17，端口 55435，避免和其他实例冲突：
docker run -d --rm --name libraryd-pg-55435 \
-e POSTGRES_DB=libraryd -e POSTGRES_USER=libraryd -e POSTGRES_PASSWORD=libraryd \
-p 127.0.0.1:55435:5432 postgres:17-alpine
until docker exec libraryd-pg-55435 pg_isready -U libraryd -d libraryd >/dev/null 2>&1; do
sleep 1
done
实验程序建三张表：books、members、loans。loans 使用第 4 章的核心列，并加上本节需要的 fine_settled_at：
create table loans (
id bigserial primary key,
isbn varchar(20) not null references books (isbn),
member_id varchar(40) not null references members (id),
borrowed_on date not null,
due_on date not null,
returned_on date,
fine_cents bigint not null default 0,
fine_settled_at timestamptz
);
生成 200 本书、100 个会员、6000 条借阅记录。借阅日期从 2025-01-01 开始按天轮转，借期 14 天；每 10 条留一条未归还，其余归还日分布在到期日前后 5 天。编译和运行：
javac -cp postgresql-42.7.14.jar -d /tmp/batch-lab BatchLab.java
java -cp /tmp/batch-lab:postgresql-42.7.14.jar BatchLab
同一台机器连续跑三次，结果如下（单位 ms）：
| 运行 | 逐条插入 | 500 条分批插入 | 逐条结算 5400 行 | 500 条分批结算 |
|---|---|---|---|---|
| 1 | 14354 | 282 | 13302 | 214 |
| 2 | 20713 | 351 | 19577 | 192 |
| 3 | 11388 | 438 | 10712 | 174 |
第三次运行的完整输出：
rows=6000 count=6000 unsettled=5400
insert row-by-row=11388ms batch(500)=438ms
settle row-by-row=10712ms rows=5400 batch=174ms rows=5400
fine total row=1540250 batch=1540250 equal=true
逐条插入在 11–21 秒之间波动，分批在 0.17–0.44 秒之间；逐条结算在 10–20 秒，分批在 0.17–0.22 秒。两次路径算出的罚金总额都是 1540250 分，说明批处理没有改变结果。
这些数字只代表这台机器、这个容器、这个数据分布。换一台磁盘更快的机器、换 synchronous_commit 设置、换 PostgreSQL 版本，绝对值都会变。要带走的是对比方法和命令：同一个 JVM、同一个数据库、同一份数据，分别跑逐条和分批，记录每次的毫秒数。
逐条插入的核心代码：
static void rowInsert(Connection c, List<LoanRow> rows) throws SQLException {
String sql = """
insert into loans (isbn, member_id, borrowed_on, due_on, returned_on)
values (?, ?, ?, ?, ?)
""";
try (PreparedStatement ps = c.prepareStatement(sql)) {
for (LoanRow r : rows) {
ps.setString(1, r.isbn());
ps.setString(2, r.memberId());
ps.setObject(3, r.borrowed());
ps.setObject(4, r.due());
ps.setObject(5, r.returned());
ps.executeUpdate();
}
}
}
分批版本把 executeUpdate 换成 addBatch，每 500 条 executeBatch 一次，并且关闭自动提交：
static void batchInsert(Connection c, List<LoanRow> rows) throws SQLException {
String sql = """
insert into loans (isbn, member_id, borrowed_on, due_on, returned_on)
values (?, ?, ?, ?, ?)
""";
c.setAutoCommit(false);
try (PreparedStatement ps = c.prepareStatement(sql)) {
int n = 0;
for (LoanRow r : rows) {
ps.setString(1, r.isbn());
ps.setString(2, r.memberId());
ps.setObject(3, r.borrowed());
ps.setObject(4, r.due());
ps.setObject(5, r.returned());
ps.addBatch();
if (++n % 500 == 0) {
ps.executeBatch();
}
}
ps.executeBatch();
c.commit();
} finally {
c.setAutoCommit(true);
}
}
setAutoCommit(false) 是收益的主要来源。批处理本身减少了网络往返和语句解析，但如果仍然是自动提交，每批还是会触发提交；把整批放在一个事务里，WAL 刷盘次数从 6000 次降到十几次。
批大小 500 是实验值，不是固定答案。太小，提交次数多；太大，内存和锁持有时间长，失败重试的代价也大。常见范围是 200–1000，最终要按行宽、数据库参数和失败恢复要求调。
游标：别把结果集一次读进内存
插入可以用 List 一次装下 6000 行，结算不行。生产环境的借阅表可能有几百万行，select * 全部读进 List 会占大量内存，还可能因为结果集太大而超时。JDBC 的 setFetchSize 可以让驱动分批从服务器取行，但有一个前提：连接必须关闭自动提交。PostgreSQL JDBC 在自动提交模式下会忽略 setFetchSize，把整个结果集读进来。
结算任务用 keyset 分页，而不是 offset：
private static final int BATCH_SIZE = 500;
private List<LoanFineRow> loadBatch(Connection c, int shard, int shards, long lastId)
throws SQLException {
String sql = """
select id, due_on, returned_on
from loans
where returned_on is not null
and fine_settled_at is null
and mod(id, ?) = ?
and id > ?
order by id
limit ?
""";
try (PreparedStatement ps = c.prepareStatement(sql)) {
ps.setInt(1, shards);
ps.setInt(2, shard);
ps.setLong(3, lastId);
ps.setInt(4, BATCH_SIZE);
ps.setFetchSize(BATCH_SIZE);
try (ResultSet rs = ps.executeQuery()) {
List<LoanFineRow> batch = new ArrayList<>(BATCH_SIZE);
while (rs.next()) {
batch.add(new LoanFineRow(
rs.getLong("id"),
rs.getObject("due_on", LocalDate.class),
rs.getObject("returned_on", LocalDate.class)));
}
return batch;
}
}
}
keyset 分页用「上一批的最大 id」作为下一页的起点，而不是 offset 500、offset 1000。原因有两个：
offset越大，数据库要跳过的行越多，复杂度接近 O(n²)。- 结算过程中行的
fine_settled_at会变化，offset分页可能重复或漏行；keyset 用稳定的 id 排序，不会因为前面的行被更新而错位。
setFetchSize(500) 和 limit 500 看起来重复，但作用不同：limit 控制一批的业务大小，fetchSize 控制驱动从服务器拉取的行数。两者取同一个值，网络往返和内存占用都比较可控。查询仍然走 ix_loans_fine_pending 索引，条件里的 fine_settled_at is null 和 id > ? 都能用上。
分片：多实例并行的边界
单实例处理 6000 行只要几百毫秒，几百万行就需要分片。最简单的分片是按 id 取模：
where mod(id, ?) = ?
每个实例负责一个 shard，所有实例的并集覆盖全表。分片数不要超过数据库连接和 CPU 能承受的并行度；4 个实例配 4 个分片，比 32 个分片更可控。
更稳妥的并发方式是 for update skip locked，让多个 worker 从同一张待处理表里抢批次，而不是预先固定分片：
select id, due_on, returned_on
from loans
where returned_on is not null
and fine_settled_at is null
order by id
limit 500
for update skip locked
skip locked 让第二个 worker 跳过已被锁定的行，继续取下一批。它适合「任务队列」式的并行，不需要预先分片；代价是锁持有到事务结束，批次不能太大。libraryd 的结算任务可以先用固定分片，等实例数变化频繁时再考虑 skip locked。
分片键要稳定。mod(id, shards) 在 id 是 bigserial 且只增不减时分布均匀；如果 id 有大量删除或手工插入，分布可能倾斜。另一种是按 id 范围分片（id between ? and ?），需要预先知道边界。无论哪种，分片数变化时都要重新分配，不能假设「分片 0 永远处理前一半」。
在 libraryd 里落地
先加迁移。fine_settled_at 是本节引入的结算标记；fine_cents = 0 不能当标记，因为按时归还会算出 0 罚金，和「未结算」无法区分。这一点在 11.3 会展开。
-- V5__fine_settlement.sql
alter table loans add column fine_settled_at timestamptz;
create index ix_loans_fine_pending
on loans (id)
where returned_on is not null and fine_settled_at is null;
FineSettlementJob 的骨架：
package com.plume.library.app;
import com.plume.library.domain.FinePolicy;
import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.time.LocalDate;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.List;
public final class FineSettlementJob {
public record SettlementReport(int scanned, int updated, int shard, int shards, long elapsedMs) {
}
private static final int BATCH_SIZE = 500;
private final DataSource dataSource;
public FineSettlementJob(DataSource dataSource) {
this.dataSource = dataSource;
}
public SettlementReport settle(int shard, int shards) {
long start = System.nanoTime();
int scanned = 0;
int updated = 0;
long lastId = 0;
try (Connection connection = dataSource.getConnection()) {
connection.setAutoCommit(false);
while (true) {
List<LoanFineRow> batch = loadBatch(connection, shard, shards, lastId);
if (batch.isEmpty()) {
connection.commit();
break;
}
updated += updateBatch(connection, batch);
connection.commit();
lastId = batch.get(batch.size() - 1).id();
scanned += batch.size();
}
} catch (SQLException e) {
throw new IllegalStateException("fine settlement failed at shard " + shard, e);
}
long elapsedMs = (System.nanoTime() - start) / 1_000_000;
return new SettlementReport(scanned, updated, shard, shards, elapsedMs);
}
private int updateBatch(Connection connection, List<LoanFineRow> batch) throws SQLException {
String sql = """
update loans
set fine_cents = ?, fine_settled_at = now()
where id = ? and fine_settled_at is null
""";
try (PreparedStatement ps = connection.prepareStatement(sql)) {
for (LoanFineRow row : batch) {
ps.setLong(1, fineCents(row));
ps.setLong(2, row.id());
ps.addBatch();
}
int[] counts = ps.executeBatch();
int updated = 0;
for (int count : counts) {
if (count > 0) {
updated += count;
}
}
return updated;
}
}
private long fineCents(LoanFineRow row) {
long days = Math.max(0, ChronoUnit.DAYS.between(row.dueOn(), row.returnedOn()));
return days * FinePolicy.dailyFineCents();
}
record LoanFineRow(long id, LocalDate dueOn, LocalDate returnedOn) {
}
}
每批提交一次，而不是整个任务一个大事务。这样做有三个好处：
- 失败时只丢当前批，前面已提交的批次不用重做。
- 锁持有时间短，不会阻塞其他写入。
- WAL 和回滚段不会因为几百万行的一次事务膨胀。
代价是任务不再「原子」：中途失败会留下「部分已结算」的状态。这没问题，因为下一轮查询用 fine_settled_at is null 会跳过已结算的行；任务要可重入，而不是要求一次成功。11.3 会把这一点和 job_runs 抢占结合起来。
update ... where id = ? and fine_settled_at is null 是第二道保险。即使两个 worker 抢到同一行，只有一个 executeUpdate 返回 1，另一个返回 0。executeBatch 的返回数组要按 > 0 统计，不能简单用数组长度，因为 SUCCESS_NO_INFO 是 -2，不能当成 1。
常见错误与边界
「自动提交模式下用 setFetchSize。」 PostgreSQL JDBC 在自动提交时会忽略 fetch size，把整个结果集读进内存。要先 setAutoCommit(false)，再执行查询；查询结束后再决定提交还是回滚。
「批大小越大越好。」 批次太大会占内存、延长锁持有时间，失败重试的成本也高。PostgreSQL 的单条 SQL 参数上限是 65535，500 条 × 5 列的插入远低于上限，但更宽的表要注意。200–1000 是常见起点，按实测调整。
「用 offset 做深分页。」 offset 100000 要让数据库跳过十万行，越往后越慢；结算过程中行还在变化，还可能重复或漏行。用 keyset（id > lastId）或 skip locked。
「整个任务一个大事务。」 几百万行的一个事务会持有大量锁，WAL 膨胀，失败回滚很慢。按批提交，配合幂等条件，失败后可以从头重跑。
「executeBatch 失败就当作全部失败。」 BatchUpdateException 里可能有部分成功的计数。记录已经成功的批次和 lastId，下一轮从断点继续；不要假设批处理是原子的。
「分片数随便改。」 分片数变化会改变 mod(id, shards) 的归属，正在跑的任务可能重复处理。分片配置要版本化，变更时先停任务，或者用 skip locked 的动态抢占替代固定分片。
「用 fine_cents = 0 判断未结算。」 按时归还会算出 0 罚金，第二次运行会把它们再处理一遍。必须用独立的 fine_settled_at 标记；11.3 会用真实输出展示这个坑。
小结
- 逐条与分批的差距主要来自提交次数：6000 行逐条插入 11–21 秒，500 条一批 0.17–0.44 秒；结算 5400 行也有同样量级的差距。
- 这些数字只用于同机对比，命令、数据分布和 PostgreSQL 版本都要一起记录。
setFetchSize必须配合setAutoCommit(false)；否则 PostgreSQL JDBC 会忽略它。- keyset 分页用
id > lastId，避免offset的深分页和行变化导致的重复/漏行。 - 每批提交一次，任务可重入；失败后靠
fine_settled_at is null跳过已处理行。 - 固定分片用
mod(id, shards)，动态抢占用for update skip locked，两者都要控制批次大小。 fine_cents = 0不能当未结算标记，必须有独立的fine_settled_at。
批处理已经能重跑，但「重跑」不等于「不会重复结算」。如果调度器触发两次、两个实例同时启动，或者任务在中途失败后重试，罚金仍可能被算两遍。下一节用数据库里的执行标记和业务键把幂等写死。
阅读导航：上一节：11.1 ScheduledExecutorService：周期任务与时钟 · 下一节：11.3 任务幂等：重复执行不出错 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。