业务系统性能改造手册 · 执行提速
把逐条访问改成批量处理
2026-08-056 min read执行提速
摘要
将订单列表中的逐条查询和循环写入改成有边界的集合读取与 JDBC Batch，在减少数据库往返的同时明确批次大小、事务、结果对应、部分失败和恢复语义。
03-01 已经把一条订单查询内部的扫描、排序、返回字段和对象装配收窄。接口仍可能慢在另一处：列表先查一批订单，随后在循环里为每个订单查询明细；批量任务也可能逐条执行更新，每次都承担一次数据库往返。
举个简化场景。一页订单已经返回若干 orderId，代码却反复调用 findItems(orderId)。单条 SQL 的执行计划都很正常，入口请求仍会产生“列表一次 + 每个订单一次”的访问。问题不在某条 SQL 特别慢，而在固定开销被重复支付。
小编对批处理的判断是：它用更大的单次输入、事务和内存占用，换取更少的往返。一次塞入全部数据不叫完成改造；输入边界、结果对应和失败恢复没有写清楚，往返成本只是换成了超长 SQL、锁等待或不可恢复的大事务。
一、先证明逐条访问发生在哪里
N+1 不只出现在 ORM 懒加载。循环里的 Mapper 调用、逐条 RPC 后再查库、保存方法内部隐含的查询，以及业务校验中的重复读取，都可能制造相同形状。
先在一次逻辑操作内连接入口、SQL 和业务键：
yaml
1rowByRowCandidate:
2 operation: ${ENDPOINT_JOB_OR_COMMAND}
3 experimentId: ${EXPERIMENT_ID}
4 inputItems: ${COUNT}
5 sqlExecutions:
6 parentQuery: ${COUNT}
7 repeatedQuery: ${COUNT}
8 repeatedUpdate: ${COUNT}
9 repeatedShape:
10 queryId: ${NORMALIZED_SQL_OR_MAPPER_ID}
11 codeLocation: ${FILE_METHOD_OR_SPAN}
12 businessKey: ${ORDER_ITEM_OR_OTHER}
13 cost:
14 databaseRoundTrips: ${COUNT_OR_ESTABLISHED_EQUIVALENT}
15 rowsReturned: ${COUNT}
16 requestP95: ${DURATION}
17 requestP99: ${DURATION}
18 connectionHoldTime: ${DURATION_OR_UNKNOWN}
19 correctness:
20 missingItemBehavior: ${EMPTY_ERROR_OR_OTHER}
21 resultOrderContract: ${INPUT_ORDER_DATABASE_ORDER_OR_NONE}
22 transactionBoundary: ${DESCRIPTION}inputItems 与重复 SQL 次数一起观察。两者随同一负载近似同步增长，才支持“逐条访问放大”的判断。某个循环内的 SQL 语义彼此不同，或者每次执行依赖上一条结果时，不能只凭代码长得像循环就合并。
本篇只处理单个业务动作内部的数据库往返。是否增加工作线程、扩大连接池或并行执行批次属于 03-03，这里保持串行且有界，避免用并发掩盖数据库容量。
二、集合读取要把对应关系补回来
把 WHERE order_id = ? 改成 WHERE order_id IN (...) 只是第一步。批量结果不会自动按输入 ID 排列；某些订单没有明细时，结果集中也不会出现占位行；输入还可能为空、重复或超过驱动与数据库可接受的参数规模。
2.1 空集合不执行 SQL，缺失项返回明确空值
下面的标准 JDBC 示例让批量入口重新检查整组订单的读取权限，tenantId 仍来自可信 AccessScope，不能由请求参数替代。列名和类型用于说明批量边界，真实 schema、JDBC 驱动和参数上限尚未确定。
java
1import java.math.BigDecimal;
2import java.sql.Connection;
3import java.sql.PreparedStatement;
4import java.sql.ResultSet;
5import java.sql.SQLException;
6import java.util.ArrayList;
7import java.util.Collections;
8import java.util.LinkedHashMap;
9import java.util.LinkedHashSet;
10import java.util.List;
11import java.util.Map;
12import java.util.Set;
13import javax.sql.DataSource;
14
15record AccessScope(long tenantId) {}
16
17record OrderItemRow(
18 long itemId,
19 long orderId,
20 String sku,
21 int quantity,
22 BigDecimal unitPrice) {}
23
24interface OrderItemBatchAuthorizer {
25 void requireReadItems(AccessScope scope, List<Long> orderIds);
26}
27
28final class OrderItemBatchReader {
29 private final DataSource dataSource;
30 private final OrderItemBatchAuthorizer authorizer;
31 private final int maxReadBatchSize;
32
33 OrderItemBatchReader(
34 DataSource dataSource,
35 OrderItemBatchAuthorizer authorizer,
36 int maxReadBatchSize) {
37 if (maxReadBatchSize <= 0) {
38 throw new IllegalArgumentException("maxReadBatchSize must be positive");
39 }
40 this.dataSource = dataSource;
41 this.authorizer = authorizer;
42 this.maxReadBatchSize = maxReadBatchSize;
43 }
44
45 Map<Long, List<OrderItemRow>> findByOrderIds(
46 AccessScope scope,
47 List<Long> requestedOrderIds) throws SQLException {
48 List<Long> orderIds = normalizeIds(requestedOrderIds);
49 authorizer.requireReadItems(scope, orderIds);
50 Map<Long, List<OrderItemRow>> grouped = new LinkedHashMap<>();
51 for (Long orderId : orderIds) {
52 grouped.put(orderId, new ArrayList<>());
53 }
54 if (orderIds.isEmpty()) {
55 return immutableResult(grouped);
56 }
57
58 try (Connection connection = dataSource.getConnection()) {
59 for (int from = 0; from < orderIds.size(); from += maxReadBatchSize) {
60 int to = Math.min(from + maxReadBatchSize, orderIds.size());
61 readChunk(connection, scope, orderIds.subList(from, to), grouped);
62 }
63 }
64 return immutableResult(grouped);
65 }
66
67 static List<Long> normalizeIds(List<Long> requestedOrderIds) {
68 if (requestedOrderIds == null) {
69 throw new IllegalArgumentException("orderIds are required");
70 }
71 Set<Long> unique = new LinkedHashSet<>();
72 for (Long orderId : requestedOrderIds) {
73 if (orderId == null || orderId <= 0) {
74 throw new IllegalArgumentException("orderId must be positive");
75 }
76 unique.add(orderId);
77 }
78 return List.copyOf(unique);
79 }
80
81 private void readChunk(
82 Connection connection,
83 AccessScope scope,
84 List<Long> orderIds,
85 Map<Long, List<OrderItemRow>> grouped) throws SQLException {
86 String placeholders = String.join(",", Collections.nCopies(orderIds.size(), "?"));
87 String sql = """
88 SELECT id, order_id, sku, quantity, unit_price
89 FROM order_items
90 WHERE tenant_id = ?
91 AND order_id IN (%s)
92 ORDER BY order_id, id
93 """.formatted(placeholders);
94
95 try (PreparedStatement statement = connection.prepareStatement(sql)) {
96 statement.setLong(1, scope.tenantId());
97 for (int i = 0; i < orderIds.size(); i++) {
98 statement.setLong(i + 2, orderIds.get(i));
99 }
100 try (ResultSet resultSet = statement.executeQuery()) {
101 while (resultSet.next()) {
102 long orderId = resultSet.getLong("order_id");
103 List<OrderItemRow> rows = grouped.get(orderId);
104 if (rows == null) {
105 throw new SQLException("database returned an unrequested orderId");
106 }
107 rows.add(new OrderItemRow(
108 resultSet.getLong("id"),
109 orderId,
110 resultSet.getString("sku"),
111 resultSet.getInt("quantity"),
112 resultSet.getBigDecimal("unit_price")));
113 }
114 }
115 }
116 }
117
118 private Map<Long, List<OrderItemRow>> immutableResult(
119 Map<Long, List<OrderItemRow>> grouped) {
120 Map<Long, List<OrderItemRow>> result = new LinkedHashMap<>();
121 grouped.forEach((orderId, rows) -> result.put(orderId, List.copyOf(rows)));
122 return Collections.unmodifiableMap(result);
123 }
124}输入先去重并保留首次出现顺序。返回 Map 预先为每个订单建立空列表，因此“订单存在但没有明细”不会与“批量代码忘了处理该 ID”混在一起；Map 的键顺序遵循输入，单个订单内的明细顺序由 SQL 的 ORDER BY order_id, id 明确规定。调用方若需要保留重复 ID，应该在这一层结果之上按原输入重新展开，不能指望数据库返回重复分组。
空集合直接返回，不拼出 IN ()。读取按 maxReadBatchSize 分段，参数上限、SQL 文本长度、结果集大小和内存峰值都因此有了调节入口。这个值不是通用常量，要结合目标驱动、MySQL 版本、每个订单的明细分布和压测结果确定。
2.2 大结果不要先全部攒进内存
示例适用于 03-01 中已经限制页大小的订单列表。若批量任务输入远大于一页，Map<Long, List<...>> 会一直保留全部明细，此时即使 SQL 分段，应用内存仍可能随总输入增长。
更大的任务应让上游按稳定业务游标产生一批 ID，完成“读取—处理—释放”后再取下一批。这里的批次同时约束 IN 参数和结果对象存活范围。不要先把全部 ID 与全部结果装进两个集合，再宣称自己做了分批。
三、批量写入先定义事务和失败语义
逐条 executeUpdate() 改成 addBatch() 与 executeBatch() 可以减少往返，但 JDBC 没有承诺“某一条失败后所有驱动都立即停止”。驱动可能停止，也可能继续处理；BatchUpdateException 的更新计数顺序对应加入批次的顺序，值可能是实际行数、SUCCESS_NO_INFO 或 EXECUTE_FAILED。Java SE 21 BatchUpdateException
下面的示例处理一个本地事务内的订单状态更新。它使用 tenant_id + id + expected_version 做乐观校验，每个命令应更新恰好一行；任何 0、大于 1、SUCCESS_NO_INFO 或 EXECUTE_FAILED 都视为无法证明整批正确，当前事务回滚。代码不适用于由 Spring/JTA 管理的外部事务，那种场景应由事务管理器负责提交与回滚。
java
1import java.sql.BatchUpdateException;
2import java.sql.Connection;
3import java.sql.PreparedStatement;
4import java.sql.SQLException;
5import java.sql.Timestamp;
6import java.time.Instant;
7import java.util.HashSet;
8import java.util.List;
9import java.util.Set;
10import javax.sql.DataSource;
11
12record AccessScope(long tenantId) {}
13
14record StatusChange(
15 long orderId,
16 long expectedVersion,
17 String nextStatus,
18 Instant changedAt) {}
19
20interface OrderStatusBatchAuthorizer {
21 void requireStatusChanges(AccessScope scope, List<StatusChange> changes);
22}
23
24enum WriteOutcome {
25 ROLLED_BACK,
26 UNKNOWN
27}
28
29final class BatchWriteException extends SQLException {
30 private final WriteOutcome outcome;
31
32 BatchWriteException(String message, WriteOutcome outcome, Throwable cause) {
33 super(message, cause);
34 this.outcome = outcome;
35 }
36
37 WriteOutcome outcome() {
38 return outcome;
39 }
40}
41
42final class OrderStatusBatchWriter {
43 private static final String UPDATE_SQL = """
44 UPDATE orders
45 SET status = ?, version = version + 1, updated_at = ?
46 WHERE tenant_id = ? AND id = ? AND version = ?
47 """;
48
49 private final DataSource dataSource;
50 private final OrderStatusBatchAuthorizer authorizer;
51 private final int maxWriteBatchSize;
52
53 OrderStatusBatchWriter(
54 DataSource dataSource,
55 OrderStatusBatchAuthorizer authorizer,
56 int maxWriteBatchSize) {
57 if (maxWriteBatchSize <= 0) {
58 throw new IllegalArgumentException("maxWriteBatchSize must be positive");
59 }
60 this.dataSource = dataSource;
61 this.authorizer = authorizer;
62 this.maxWriteBatchSize = maxWriteBatchSize;
63 }
64
65 int updateOneChunk(AccessScope scope, List<StatusChange> changes)
66 throws SQLException {
67 validate(changes);
68 authorizer.requireStatusChanges(scope, changes);
69 if (changes.isEmpty()) {
70 return 0;
71 }
72
73 try (Connection connection = dataSource.getConnection()) {
74 boolean originalAutoCommit = connection.getAutoCommit();
75 if (!originalAutoCommit) {
76 throw new SQLException("example requires a locally managed connection");
77 }
78 connection.setAutoCommit(false);
79 boolean transactionEnded = false;
80 try {
81 int[] counts = executeBatch(connection, scope, changes);
82 requireExactlyOneRowPerChange(counts, changes.size());
83 try {
84 connection.commit();
85 } catch (SQLException commitFailure) {
86 throw new BatchWriteException(
87 "commit outcome requires reconciliation",
88 WriteOutcome.UNKNOWN,
89 commitFailure);
90 }
91 transactionEnded = true;
92 return changes.size();
93 } catch (BatchWriteException knownFailure) {
94 if (knownFailure.outcome() == WriteOutcome.UNKNOWN) {
95 throw knownFailure;
96 }
97 rollbackOrThrowUnknown(connection, knownFailure);
98 transactionEnded = true;
99 throw knownFailure;
100 } catch (SQLException | RuntimeException failure) {
101 rollbackOrThrowUnknown(connection, failure);
102 transactionEnded = true;
103 throw new BatchWriteException(
104 "batch rolled back",
105 WriteOutcome.ROLLED_BACK,
106 failure);
107 } finally {
108 if (transactionEnded) {
109 try {
110 connection.setAutoCommit(originalAutoCommit);
111 } catch (SQLException restoreFailure) {
112 // 事务已结束；连接关闭时由池适配层重置或丢弃。
113 }
114 }
115 }
116 }
117 }
118
119 private int[] executeBatch(
120 Connection connection,
121 AccessScope scope,
122 List<StatusChange> changes) throws SQLException {
123 try (PreparedStatement statement = connection.prepareStatement(UPDATE_SQL)) {
124 for (StatusChange change : changes) {
125 statement.setString(1, change.nextStatus());
126 statement.setTimestamp(2, Timestamp.from(change.changedAt()));
127 statement.setLong(3, scope.tenantId());
128 statement.setLong(4, change.orderId());
129 statement.setLong(5, change.expectedVersion());
130 statement.addBatch();
131 }
132 try {
133 return statement.executeBatch();
134 } catch (BatchUpdateException partialFailure) {
135 throw new BatchWriteException(
136 "driver reported partial batch execution",
137 WriteOutcome.ROLLED_BACK,
138 partialFailure);
139 }
140 }
141 }
142
143 static void requireExactlyOneRowPerChange(int[] counts, int expectedSize)
144 throws BatchWriteException {
145 if (counts.length != expectedSize) {
146 throw new BatchWriteException(
147 "batch result length is ambiguous",
148 WriteOutcome.ROLLED_BACK,
149 null);
150 }
151 for (int count : counts) {
152 if (count != 1) {
153 throw new BatchWriteException(
154 "expected one changed row but got " + count,
155 WriteOutcome.ROLLED_BACK,
156 null);
157 }
158 }
159 }
160
161 private void validate(List<StatusChange> changes) {
162 if (changes == null || changes.size() > maxWriteBatchSize) {
163 throw new IllegalArgumentException("invalid write batch size");
164 }
165 Set<Long> seen = new HashSet<>();
166 for (StatusChange change : changes) {
167 if (change == null
168 || change.orderId() <= 0
169 || change.expectedVersion() < 0
170 || change.nextStatus() == null
171 || change.nextStatus().isBlank()
172 || change.changedAt() == null
173 || !seen.add(change.orderId())) {
174 throw new IllegalArgumentException("invalid or duplicate status change");
175 }
176 }
177 }
178
179 private void rollbackOrThrowUnknown(Connection connection, Throwable failure)
180 throws BatchWriteException {
181 try {
182 connection.rollback();
183 } catch (SQLException rollbackFailure) {
184 failure.addSuppressed(rollbackFailure);
185 throw new BatchWriteException(
186 "rollback failed; outcome requires reconciliation",
187 WriteOutcome.UNKNOWN,
188 failure);
189 }
190 }
191}SUCCESS_NO_INFO 表示命令成功但驱动没有给出影响行数；对于要求每条乐观更新恰好命中一行的业务，它仍然不足以证明版本校验成立，因此示例选择回滚。Oracle 的 JDBC API 也明确区分实际更新数、SUCCESS_NO_INFO 与 EXECUTE_FAILED。Java SE 21 Statement
提交异常和回滚异常被标成 UNKNOWN。这条路径也不会调用 setAutoCommit(true)：JDBC 连接仍有未结束事务时，切回自动提交可能提交当前事务，反而改变不确定结果。连接关闭后的重置、销毁与告警需要由数据源适配层验证。调用方应按批次 ID、订单 ID 和期望版本查询最终状态后再决定补偿，不能看到异常就整批盲重试。当前示例依靠版本条件阻止同一版本重复更新；若写操作会产生扣款、发消息等额外副作用，还需要稳定操作 ID、唯一约束或业务状态机，不能把 JDBC Batch 当成幂等机制。
示例要求调用方先按 maxWriteBatchSize 切块，每块一个本地事务，并要求目标表使用能够兑现回滚语义的事务型存储引擎。批次越大，提交次数可能减少，但锁持有、日志量、失败回滚范围和连接占用也会扩大。跨批次不具备整体原子性：前一批提交、后一批失败时，恢复流程必须接受并记录这种状态，或者业务就不应采用分块提交。
四、批次大小只能从实验中选择
没有真实 schema、驱动配置、参数限制和数据分布，无法给出“每批多少条”的答案。读取和写入也不应共用一个数字：读取受 IN 参数、单行展开数量和结果内存影响；写入还受事务日志、锁竞争、驱动是否重写批次以及单条语句大小影响。
固定其余条件，分别测试多个候选批次，记录原始值而不是只报最快的一档：
yaml
1batchExperiment:
2 experimentId: ${EXPERIMENT_ID}
3 databaseAndDriver:
4 productVersion: ${DATABASE_VERSION}
5 jdbcDriverVersion: ${DRIVER_VERSION}
6 driverBatchOptions: ${OPTIONS}
7 workload:
8 operation: ${READ_ITEMS_OR_UPDATE_STATUS}
9 totalInputItems: ${COUNT}
10 distribution: ${ITEMS_PER_ORDER_OR_UPDATE_SHAPE}
11 batchSize: ${COUNT}
12 transactionPerBatch: ${TRUE_OR_FALSE_WITH_BOUNDARY}
13 result:
14 sqlExecutions: ${COUNT}
15 databaseRoundTrips: ${COUNT_OR_MEASURED_EQUIVALENT}
16 completedItems: ${COUNT}
17 throughput: ${VALUE}
18 requestP95: ${DURATION}
19 requestP99: ${DURATION}
20 connectionHoldTime: ${DURATION}
21 lockWait: ${DURATION_OR_UNKNOWN}
22 transactionLogEvidence: ${VALUE_OR_UNKNOWN}
23 applicationPeakMemory: ${BYTES_OR_UNKNOWN}
24 databaseCpuAndIo: ${REFERENCES}
25 terminalCounts: ${COUNTS}
26 failureInjection:
27 failedItemPosition: ${FIRST_MIDDLE_LAST_OR_NONE}
28 driverContinuedAfterFailure: ${TRUE_FALSE_OR_UNKNOWN}
29 committedChunks: ${COUNT}
30 rolledBackChunks: ${COUNT}
31 unknownOutcomeChunks: ${COUNT}
32 recoveryResult: ${REFERENCE}
33 decision:
34 selectedBatchSize: ${COUNT}
35 reason: ${MEASURED_TRADEOFF}
36 upperGuardrail: ${COUNT_AND_REASON}
37 rollbackTrigger: ${CONDITION}吞吐提高但 P99、锁等待或内存峰值失控，不算可接受结果。驱动升级或批处理参数变化后也要重测；JDBC 只定义接口语义，是否合并网络包、是否改写为多值语句以及超时作用于单条还是整批，需要查看目标驱动文档并用实际调用证据确认。
五、恢复检查表要能回答“哪些已经生效”
批量改造上线前，至少把下面的检查项变成可执行测试或操作记录：
yaml
1batchRecoveryChecklist:
2 identity:
3 batchId: ${STABLE_ID}
4 inputSnapshot: ${ID_VERSION_OR_HASH_REFERENCE}
5 readPath:
6 emptyInputRunsNoSql: ${PASS_FAIL}
7 duplicateInputContract: ${PASS_FAIL}
8 missingRowsBecomeEmptyGroups: ${PASS_FAIL}
9 resultOrderMatchesContract: ${PASS_FAIL}
10 oversizedInputIsChunked: ${PASS_FAIL}
11 writePath:
12 duplicateIdsRejected: ${PASS_FAIL}
13 zeroRowOptimisticUpdateRejected: ${PASS_FAIL}
14 successNoInfoPolicyVerified: ${PASS_FAIL}
15 middleItemFailureInjected: ${PASS_FAIL}
16 rollbackVerified: ${PASS_FAIL}
17 commitFailureOutcomeReconciled: ${PASS_FAIL}
18 recovery:
19 committedChunksLocated: ${QUERY_OR_LOG_REFERENCE}
20 unknownChunksReconciled: ${QUERY_OR_LOG_REFERENCE}
21 retryPreconditionsChecked: ${PASS_FAIL}
22 compensationAudited: ${REFERENCE}回退时先切回旧的数据访问实现，保留批次 ID、失败输入和已提交分块记录，直到不确定批次全部核对完成。不要因为旧代码恢复了就删除恢复证据；跨批次已提交的数据不会随代码开关自动回滚。
批量处理成立的证据很朴素：相同业务输入下，数据库往返与 SQL 执行次数减少，吞吐或延迟得到可重复改善，同时结果对应、事务边界、锁等待、内存和失败恢复仍符合约定。下一篇再基于这些已经测出的服务时间和连接占用，讨论线程池与连接池如何共享容量预算。