perf: 消除保存待办时的 getByTodoKey N+1，改为批量 findByKeys 分块加载 issue #210

- FlowRecordSaveService.saveTodoMargeRecords 先批量加载已存在待办，替代循环内逐条 getByTodoKey（每 20 条待办 20 次 SELECT → 1 次 IN 查询）
- 按 TODO_KEY_BATCH_SIZE=500 分块查询，避免海量 key 拼超大 IN 子句/结果集过大导致 OOM
- FlowTodoRecordRepository 新增 findByKeys，实现于 mock/test/Infra 三层仓储
- 新增集成断言：批量场景逐条 getByTodoKey 调用数远小于待办数，且走批量 findByKeys
- 全量 clean install BUILD SUCCESS，227 framework + example 真实 JPA 集成测试全绿

#210

Co-Authored-By: Claude <noreply@anthropic.com>

Author: xlorne

flow-engine-framework/src/main/java/com/codingapi/flow/mock/repository/FlowTodoRecordRepositoryMockImpl.java

@@ -4,15 +4,21 @@
 import com.codingapi.flow.repository.FlowTodoRecordRepository;
 
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 public class FlowTodoRecordRepositoryMockImpl implements FlowTodoRecordRepository {
 
     private final Map<Long, FlowTodoRecord> cache = new HashMap<>();
     private final Map<String, FlowTodoRecord> cacheByMageKey = new HashMap<>();
     private long nextId = 1;
 
+    // 查询计数器，供测试断言"按 key 逐条查询已被批量查询替代"（N+1 消除）
+    private int getByTodoKeyCalls;
+    private int findByKeysCalls;
+
     @Override
     public void save(FlowTodoRecord record) {
         if (record.getId() > 0) {
@@ -41,9 +47,28 @@ public void delete(FlowTodoRecord margeRecord) {
 
     @Override
     public FlowTodoRecord getByTodoKey(String key) {
+        getByTodoKeyCalls++;
         return cacheByMageKey.get(key);
     }
 
+    @Override
+    public List<FlowTodoRecord> findByKeys(List<String> keys) {
+        findByKeysCalls++;
+        Set<String> keySet = new HashSet<>(keys);
+        return cacheByMageKey.entrySet().stream()
+                .filter(entry -> keySet.contains(entry.getKey()))
+                .map(Map.Entry::getValue)
+                .toList();
+    }
+
+    public int getGetByTodoKeyCalls() {
+        return getByTodoKeyCalls;
+    }
+
+    public int getFindByKeysCalls() {
+        return findByKeysCalls;
+    }
+
     public List<FlowTodoRecord> findByOperatorId(long operatorId) {
         return cache.values().stream().filter(record -> record.getCurrentOperatorId() == operatorId).toList();
     }

flow-engine-framework/src/main/java/com/codingapi/flow/repository/FlowTodoRecordRepository.java

@@ -10,6 +10,14 @@ public interface FlowTodoRecordRepository {
 
     FlowTodoRecord getByTodoKey(String key);
 
+    /**
+     * 按多个待办合并 key 批量加载已存在的待办记录，用于替代循环内逐条 {@link #getByTodoKey(String)} 的 N+1 查询。
+     *
+     * @param keys 待办合并 key 列表
+     * @return 已存在的待办记录
+     */
+    List<FlowTodoRecord> findByKeys(List<String> keys);
+
     void delete(FlowTodoRecord margeRecord);
 
     void save(FlowTodoRecord margeRecord);

flow-engine-framework/src/main/java/com/codingapi/flow/service/FlowRecordSaveService.java

@@ -8,13 +8,21 @@
 import com.codingapi.flow.repository.FlowTodoRecordRepository;
 
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.Objects;
 
 /**
  * 流程记录保存服务,负责保存流程记录和待办记录的合并关系
  */
 class FlowRecordSaveService {
 
+    /**
+     * 批量加载已存在待办时单批查询的 key 上限，避免单个 IN 子句过大（SQL 限制）与结果集过大（内存）。
+     */
+    private static final int TODO_KEY_BATCH_SIZE = 500;
+
     private final List<FlowRecord> flowRecords;
 
     private FlowTodoRecordRepository flowTodoRecordRepository;
@@ -41,12 +49,29 @@ public void registerRepositories(FlowTodoRecordRepository flowTodoRecordReposito
 
 
     private void saveTodoMargeRecords() {
+        // 批量加载已存在的待办（而非逐条 getByTodoKey 的 N+1）：
+        // 按分块大小分批 findByKeys，避免海量 key 拼在单个 IN 子句里导致 SQL/结果集过大（OOM 风险）。
+        List<String> todoKeys = flowRecords.stream()
+                .filter(FlowRecord::isTodo)
+                .map(FlowRecord::getTodoKey)
+                .filter(Objects::nonNull)
+                .distinct()
+                .toList();
+        Map<String, FlowTodoRecord> existedByKey = new HashMap<>();
+        for (int start = 0; start < todoKeys.size(); start += TODO_KEY_BATCH_SIZE) {
+            List<String> chunk = todoKeys.subList(start, Math.min(start + TODO_KEY_BATCH_SIZE, todoKeys.size()));
+            for (FlowTodoRecord existed : flowTodoRecordRepository.findByKeys(chunk)) {
+                existedByKey.put(existed.getTodoKey(), existed);
+            }
+        }
+
         List<FlowTodoRecord> flowTodoRecords = new ArrayList<>();
         for (FlowRecord flowRecord : flowRecords) {
             if (flowRecord.isTodo()) {
-                FlowTodoRecord todoMargeRecord = flowTodoRecordRepository.getByTodoKey(flowRecord.getTodoKey());
+                FlowTodoRecord todoMargeRecord = existedByKey.get(flowRecord.getTodoKey());
                 if (todoMargeRecord == null) {
                     todoMargeRecord = new FlowTodoRecord(flowRecord);
+                    existedByKey.put(flowRecord.getTodoKey(), todoMargeRecord);
                 } else {
                     todoMargeRecord.update(flowRecord);
                     if (flowRecord.isMergeable()) {

flow-engine-framework/src/test/java/com/codingapi/flow/repository/FlowTodoRecordRepositoryImpl.java

@@ -3,15 +3,21 @@
 import com.codingapi.flow.record.FlowTodoRecord;
 
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 
 public class FlowTodoRecordRepositoryImpl implements FlowTodoRecordRepository {
 
     private final Map<Long, FlowTodoRecord> cache = new HashMap<>();
     private final Map<String, FlowTodoRecord> cacheByMageKey = new HashMap<>();
     private long nextId = 1;
 
+    // 查询计数器，供测试断言"按 key 逐条查询已被批量查询替代"（N+1 消除）
+    private int getByTodoKeyCalls;
+    private int findByKeysCalls;
+
     @Override
     public void save(FlowTodoRecord record) {
         if (record.getId() > 0) {
@@ -40,9 +46,28 @@ public void delete(FlowTodoRecord margeRecord) {
 
     @Override
     public FlowTodoRecord getByTodoKey(String key) {
+        getByTodoKeyCalls++;
         return cacheByMageKey.get(key);
     }
 
+    @Override
+    public List<FlowTodoRecord> findByKeys(List<String> keys) {
+        findByKeysCalls++;
+        Set<String> keySet = new HashSet<>(keys);
+        return cacheByMageKey.entrySet().stream()
+                .filter(entry -> keySet.contains(entry.getKey()))
+                .map(Map.Entry::getValue)
+                .toList();
+    }
+
+    public int getGetByTodoKeyCalls() {
+        return getByTodoKeyCalls;
+    }
+
+    public int getFindByKeysCalls() {
+        return findByKeysCalls;
+    }
+
     public List<FlowTodoRecord> findByOperatorId(long operatorId) {
         return cache.values().stream().filter(record -> record.getCurrentOperatorId() == operatorId).toList();
     }

flow-engine-framework/src/test/java/com/codingapi/flow/service/FlowIssue210SubProcessPerformanceTest.java

@@ -39,6 +39,7 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 /**
  * 问题 210 复现测试：主流程 C（子流程节点）一次性创建 6 个子流程、每个子流程 B 节点 20 个审批人。
@@ -127,6 +128,30 @@ void shouldMeasureSubProcessBulkCreationTimeWithGatewayDelay() {
                 + " (网关延时 " + GATEWAY_DELAY_MS + "ms/次)");
     }
 
+    /**
+     * 测试目标：验证保存待办时不再逐条 getByTodoKey（N+1），而是改用批量 findByKeys。
+     * 前置条件：网关不带延时（本类每次测试都新建 factory），强制执行 bulk 场景。
+     * 执行步骤：完成与 {@code shouldMeasureSubProcessBulkCreationTimeWithGatewayDelay} 相同的
+     *           6 子流程 × 20 审批人（120 条待办）创建。
+     * 期望断言：逐条 getByTodoKey 的调用次数远小于待办数量（仅来自节点完成清理路径），
+     *           且调用过批量 findByKeys。
+     */
+    @Test
+    void shouldBatchLoadExistingTodosInsteadOfPerRecordGet() {
+        Map<String, Object> data = new LinkedHashMap<>();
+        data.put("content", "parent");
+        data.put("approvers", approvers.stream().map(User::getUserId).toList());
+        long parentRecordId = createParent(data);
+        approveMain(factory.flowRecordRepository.get(parentRecordId), parentStart, data);
+        FlowRecord parentBRecord = findTodo(initiator, parentB.getId());
+        approveMain(parentBRecord, parentB, data);
+
+        assertTrue(factory.flowTodoRecordRepository.getGetByTodoKeyCalls() < APPROVER_COUNT,
+                "批量创建待办时应以批量 findByKeys 替代逐条 getByTodoKey，避免 N+1");
+        assertTrue(factory.flowTodoRecordRepository.getFindByKeysCalls() >= 1,
+                "应调用批量 findByKeys 加载已存在待办");
+    }
+
     // ---------- 网关延时 + 计数 ----------
 
     private static class DelayedGateway implements FlowOperatorGateway {

flow-engine-starter-infra/src/main/java/com/codingapi/flow/infra/jpa/FlowTodoRecordEntityRepository.java

@@ -6,10 +6,15 @@
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.jpa.repository.Query;
 
+import java.util.List;
+
 public interface FlowTodoRecordEntityRepository extends FastRepository<FlowTodoRecordEntity,Long> {
 
     FlowTodoRecordEntity getByTodoKey(String todoKey);
 
+    @Query("from FlowTodoRecordEntity r where r.todoKey in ?1")
+    List<FlowTodoRecordEntity> findByKeys(List<String> keys);
+
 
     @Query("from FlowTodoRecordEntity r where r.currentOperatorId = ?1")
     Page<FlowTodoRecordEntity> findTodoRecordPage(long currentOperatorId, PageRequest pageRequest);

flow-engine-starter-infra/src/main/java/com/codingapi/flow/infra/repository/impl/FlowTodoRecordRepositoryImpl.java

@@ -26,6 +26,13 @@ public FlowTodoRecord getByTodoKey(String key) {
         return FlowTodoRecordConvertor.convert(flowTodoRecordEntityRepository.getByTodoKey(key));
     }
 
+    @Override
+    public List<FlowTodoRecord> findByKeys(List<String> keys) {
+        return flowTodoRecordEntityRepository.findByKeys(keys).stream()
+                .map(FlowTodoRecordConvertor::convert)
+                .toList();
+    }
+
     @Override
     public void delete(FlowTodoRecord margeRecord) {
         flowTodoRecordEntityRepository.deleteById(margeRecord.getId());