super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(similar-asin): 抽出 LlmPipelineSupport，Service 降至 1735 行（累计 -70.7%）

Browse Source 
在上一提交（3455 行）基础上，把「Python 结果回传 → 分片落库 → LLM 检测」整条流水线
（57 个方法 / 1694 行）抽为 SimilarAsinPipelineSupport。等价搬移，未改行为。

采用依赖倒置消除循环依赖：support 包声明 SimilarAsinPipelineHost 接口
（finalizeTask / findOrCreateResultRecordForAssembly / readCategorySwitch /
finalizeExhaustedResultFileJob），由 SimilarAsinTaskService 实现；这几项保留在宿主
是因为它们属于编排与事务边界（handleResultFileJobFailure 带 @Transactional）。

同时把 5 个被两侧共用的内部 record 提为 support 包顶层类型
（SubmittedTaskMetadata / FinalizeTaskResult / SubmitContext /
 PersistSubmittedChunkResult / PreparedSubmittedChunk），3 个仅流水线内部使用的
record 内联进流水线类；LlmBatchContext 一并归位。

测试适配：4 个测试类里对 mergeLlmRowsIntoChunk / bufferLlmRowsOrMerge /
flushLlmBufferedResults 的反射改指向流水线实例；RollbackSemanticsContractTest
通过 pipelineSupport() 反射取得实例（跨包，保持封装）。

验证：干净工作区叠加本改动跑 similarasin 386 + task 引用方 166 测试，
结果与基线一致（仅既有失败），零新增失败。
This commit is contained in:
黄自达
2026-09-14 01:51:36 +08:00
parent 9d92fb4af2
commit b05bba50fa
13 changed files with 2105 additions and 1810 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceChunkMergeLimitTest.java
+3 -2
View File
@@ -20,6 +20,7 @@ import org.apache.ibatis.builder.MapperBuilderAssistant;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import com.nanri.aiimage.modules.similarasin.service.support.SimilarAsinPipelineSupport;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
@@ -166,10 +167,10 @@ class SimilarAsinTaskServiceChunkMergeLimitTest {
private static void invokeMerge(SimilarAsinTaskService service, FileTaskEntity task,
String scopeHash, Integer chunkIndex,
List<SimilarAsinResultRowDto> llmRows) throws Exception {
Method merge = SimilarAsinTaskService.class.getDeclaredMethod("mergeLlmRowsIntoChunk",
Method merge = SimilarAsinPipelineSupport.class.getDeclaredMethod("mergeLlmRowsIntoChunk",
FileTaskEntity.class, String.class, Integer.class, List.class, Map.class);
merge.setAccessible(true);
merge.invoke(service, task, scopeHash, chunkIndex, llmRows, Map.of());
merge.invoke(service.pipelineSupport(), task, scopeHash, chunkIndex, llmRows, Map.of());
}
@Test
backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceCozeBufferScopeTest.java
+20 -19
View File
@@ -23,6 +23,7 @@ import org.apache.ibatis.builder.MapperBuilderAssistant;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import com.nanri.aiimage.modules.similarasin.service.support.SimilarAsinPipelineSupport;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
@@ -226,12 +227,12 @@ class SimilarAsinTaskServiceCozeBufferScopeTest {
FileTaskEntity task = task();
List<SimilarAsinResultRowDto> rows = List.of(row("r1", "1", "B0A0000001", "英国", "Title 1"));
Method bufferOrMerge = SimilarAsinTaskService.class.getDeclaredMethod("bufferLlmRowsOrMerge",
Method bufferOrMerge = SimilarAsinPipelineSupport.class.getDeclaredMethod("bufferLlmRowsOrMerge",
TaskScopeStateEntity.class,
LlmBatchContext.class,
List.class, FileTaskEntity.class, Map.class);
bufferOrMerge.setAccessible(true);
bufferOrMerge.invoke(service, state(task, 1L, "DONE", 1), context(1), rows, task, Map.of());
bufferOrMerge.invoke(service.pipelineSupport(), state(task, 1L, "DONE", 1), context(1), rows, task, Map.of());
verify(transientPayloadStorageService, times(1)).storeParsedPayloadEntry(
eq(MODULE), eq(7104L), anyString(), anyString(), anyString(), eq(true));
@@ -264,9 +265,9 @@ class SimilarAsinTaskServiceCozeBufferScopeTest {
when(taskChunkMapper.selectList(any())).thenReturn(List.of(chunk(1L, "scope-1", 1, "ptr:chunk-1")));
when(taskChunkMapper.update(any(), any())).thenReturn(1);
Method flush = SimilarAsinTaskService.class.getDeclaredMethod("flushLlmBufferedResults", Long.class);
Method flush = SimilarAsinPipelineSupport.class.getDeclaredMethod("flushLlmBufferedResults", Long.class);
flush.setAccessible(true);
flush.invoke(service, 7104L);
flush.invoke(service.pipelineSupport(), 7104L);
verify(transientPayloadStorageService, atLeastOnce()).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString());
verify(taskChunkMapper, atLeastOnce()).update(any(), any());
@@ -280,14 +281,14 @@ class SimilarAsinTaskServiceCozeBufferScopeTest {
FileTaskEntity task = task();
List<SimilarAsinResultRowDto> rows = List.of(row("r1", "1", "B0A0000001", "英国", "Title 1"));
Method bufferOrMerge = SimilarAsinTaskService.class.getDeclaredMethod("bufferLlmRowsOrMerge",
Method bufferOrMerge = SimilarAsinPipelineSupport.class.getDeclaredMethod("bufferLlmRowsOrMerge",
TaskScopeStateEntity.class,
LlmBatchContext.class,
List.class, FileTaskEntity.class, Map.class);
bufferOrMerge.setAccessible(true);
TaskScopeStateEntity state = state(task, 1L, "DONE", 2);
bufferOrMerge.invoke(service, state, context(2), rows, task, Map.of());
bufferOrMerge.invoke(service, state, context(2), rows, task, Map.of());
bufferOrMerge.invoke(service.pipelineSupport(), state, context(2), rows, task, Map.of());
bufferOrMerge.invoke(service.pipelineSupport(), state, context(2), rows, task, Map.of());
// 缓冲 2 次（每次重新写 pointer 是幂等语义：同一 state 覆盖写，无重复行）
verify(transientPayloadStorageService, times(2)).storeParsedPayloadEntry(
@@ -299,13 +300,13 @@ class SimilarAsinTaskServiceCozeBufferScopeTest {
void test_task_012_payload_chunk_boundary_empty_input() throws Exception {
// 空输入：无行时缓冲与 merge 都不发生，不创建无效资源
FileTaskEntity task = task();
Method bufferOrMerge = SimilarAsinTaskService.class.getDeclaredMethod("bufferLlmRowsOrMerge",
Method bufferOrMerge = SimilarAsinPipelineSupport.class.getDeclaredMethod("bufferLlmRowsOrMerge",
TaskScopeStateEntity.class,
LlmBatchContext.class,
List.class, FileTaskEntity.class, Map.class);
bufferOrMerge.setAccessible(true);
bufferOrMerge.invoke(service, state(task, 1L, "DONE", 1), context(1), null, task, Map.of());
bufferOrMerge.invoke(service, state(task, 2L, "DONE", 1), context(1), List.of(), task, Map.of());
bufferOrMerge.invoke(service.pipelineSupport(), state(task, 1L, "DONE", 1), context(1), null, task, Map.of());
bufferOrMerge.invoke(service.pipelineSupport(), state(task, 2L, "DONE", 1), context(1), List.of(), task, Map.of());
verify(transientPayloadStorageService, never()).storeParsedPayloadEntry(any(), any(), anyString(), anyString(), anyString(), eq(true));
verify(taskChunkMapper, never()).update(any(), any());
@@ -316,12 +317,12 @@ class SimilarAsinTaskServiceCozeBufferScopeTest {
// 单 batch（batchTotal=1）：原 P0-3 例外，现在也缓冲
FileTaskEntity task = task();
List<SimilarAsinResultRowDto> rows = List.of(row("r1", "1", "B0A0000001", "英国", "Title 1"));
Method bufferOrMerge = SimilarAsinTaskService.class.getDeclaredMethod("bufferLlmRowsOrMerge",
Method bufferOrMerge = SimilarAsinPipelineSupport.class.getDeclaredMethod("bufferLlmRowsOrMerge",
TaskScopeStateEntity.class,
LlmBatchContext.class,
List.class, FileTaskEntity.class, Map.class);
bufferOrMerge.setAccessible(true);
bufferOrMerge.invoke(service, state(task, 1L, "DONE", 1), context(1), rows, task, Map.of());
bufferOrMerge.invoke(service.pipelineSupport(), state(task, 1L, "DONE", 1), context(1), rows, task, Map.of());
verify(transientPayloadStorageService, times(1)).storeParsedPayloadEntry(
eq(MODULE), eq(7104L), anyString(), anyString(), anyString(), eq(true));
@@ -335,12 +336,12 @@ class SimilarAsinTaskServiceCozeBufferScopeTest {
stubChunkMerge(chunkRowsJson());
when(properties.isLlmResultBufferEnabled()).thenReturn(false);
List<SimilarAsinResultRowDto> rows = List.of(row("r1", "1", "B0A0000001", "英国", "Title 1"));
Method bufferOrMerge = SimilarAsinTaskService.class.getDeclaredMethod("bufferLlmRowsOrMerge",
Method bufferOrMerge = SimilarAsinPipelineSupport.class.getDeclaredMethod("bufferLlmRowsOrMerge",
TaskScopeStateEntity.class,
LlmBatchContext.class,
List.class, FileTaskEntity.class, Map.class);
bufferOrMerge.setAccessible(true);
bufferOrMerge.invoke(service, state(task, 1L, "DONE", 1), context(1), rows, task, Map.of());
bufferOrMerge.invoke(service.pipelineSupport(), state(task, 1L, "DONE", 1), context(1), rows, task, Map.of());
verify(transientPayloadStorageService, never()).storeParsedPayloadEntry(any(), any(), anyString(), anyString(), anyString(), eq(true));
verify(transientPayloadStorageService, atLeastOnce()).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString());
@@ -355,12 +356,12 @@ class SimilarAsinTaskServiceCozeBufferScopeTest {
eq(MODULE), eq(7104L), anyString(), anyString(), anyString(), eq(true)))
.thenThrow(new IllegalStateException("rustfs full"));
List<SimilarAsinResultRowDto> rows = List.of(row("r1", "1", "B0A0000001", "英国", "Title 1"));
Method bufferOrMerge = SimilarAsinTaskService.class.getDeclaredMethod("bufferLlmRowsOrMerge",
Method bufferOrMerge = SimilarAsinPipelineSupport.class.getDeclaredMethod("bufferLlmRowsOrMerge",
TaskScopeStateEntity.class,
LlmBatchContext.class,
List.class, FileTaskEntity.class, Map.class);
bufferOrMerge.setAccessible(true);
bufferOrMerge.invoke(service, state(task, 1L, "DONE", 1), context(1), rows, task, Map.of());
bufferOrMerge.invoke(service.pipelineSupport(), state(task, 1L, "DONE", 1), context(1), rows, task, Map.of());
verify(transientPayloadStorageService, atLeastOnce()).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString());
verify(taskChunkMapper, atLeastOnce()).update(any(), any());
@@ -394,11 +395,11 @@ class SimilarAsinTaskServiceCozeBufferScopeTest {
return "ptr:stored-" + invocation.getArgument(3);
}).when(transientPayloadStorageService).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString());
Method flush = SimilarAsinTaskService.class.getDeclaredMethod("flushLlmBufferedResults", Long.class);
Method flush = SimilarAsinPipelineSupport.class.getDeclaredMethod("flushLlmBufferedResults", Long.class);
flush.setAccessible(true);
Exception ex = assertThrows(Exception.class, () -> {
try {
flush.invoke(service, 7104L);
flush.invoke(service.pipelineSupport(), 7104L);
} catch (java.lang.reflect.InvocationTargetException e) {
throw e.getCause();
}
@@ -410,7 +411,7 @@ class SimilarAsinTaskServiceCozeBufferScopeTest {
verify(taskScopeStateMapper, never()).update(any(), any());
// 恢复后重试 flush：chunk 合并成功一次，pointer 清理
flush.invoke(service, 7104L);
flush.invoke(service.pipelineSupport(), 7104L);
assertEquals(2, storeCalls.get(), "恢复后重试应再次写 chunk");
verify(taskScopeStateMapper, atLeastOnce()).update(any(), any());
}
backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceRowKeyDedupeTest.java
+8 -7
View File
@@ -20,6 +20,7 @@ import org.apache.ibatis.builder.MapperBuilderAssistant;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import com.nanri.aiimage.modules.similarasin.service.support.SimilarAsinPipelineSupport;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
@@ -171,10 +172,10 @@ class SimilarAsinTaskServiceRowKeyDedupeTest {
assertEquals(1, deduped.size(), "重复行必须按稳定 rowKey 去重");
assertEquals("r1", deduped.get(0).getRowToken());
Method merge = SimilarAsinTaskService.class.getDeclaredMethod("mergeLlmRowsIntoChunk",
Method merge = SimilarAsinPipelineSupport.class.getDeclaredMethod("mergeLlmRowsIntoChunk",
FileTaskEntity.class, String.class, Integer.class, List.class, Map.class);
merge.setAccessible(true);
merge.invoke(service, task, null, null, llmRows, Map.of());
merge.invoke(service.pipelineSupport(), task, null, null, llmRows, Map.of());
assertEquals(1, storedCounter.get(), "去重后 chunk 只写一次");
verify(transientPayloadStorageService, times(1)).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString());
}
@@ -208,10 +209,10 @@ class SimilarAsinTaskServiceRowKeyDedupeTest {
FileTaskEntity task = new FileTaskEntity();
task.setId(7004L);
Method merge = SimilarAsinTaskService.class.getDeclaredMethod("mergeLlmRowsIntoChunk",
Method merge = SimilarAsinPipelineSupport.class.getDeclaredMethod("mergeLlmRowsIntoChunk",
FileTaskEntity.class, String.class, Integer.class, List.class, Map.class);
merge.setAccessible(true);
merge.invoke(service, task, null, null, llmRows, Map.of());
merge.invoke(service.pipelineSupport(), task, null, null, llmRows, Map.of());
verify(transientPayloadStorageService, times(2)).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString());
}
@@ -297,12 +298,12 @@ class SimilarAsinTaskServiceRowKeyDedupeTest {
.thenThrow(new IllegalStateException("rustfs down"));
FileTaskEntity task = new FileTaskEntity();
task.setId(7004L);
Method merge = SimilarAsinTaskService.class.getDeclaredMethod("mergeLlmRowsIntoChunk",
Method merge = SimilarAsinPipelineSupport.class.getDeclaredMethod("mergeLlmRowsIntoChunk",
FileTaskEntity.class, String.class, Integer.class, List.class, Map.class);
merge.setAccessible(true);
Exception ex = assertThrows(Exception.class, () -> {
try {
merge.invoke(service, task, null, null,
merge.invoke(service.pipelineSupport(), task, null, null,
List.of(row("r1", "1", "B0A0000001", "英国"), row("r1", "1", "B0A0000001", "英国")), Map.of());
} catch (java.lang.reflect.InvocationTargetException e) {
throw e.getCause();
@@ -314,7 +315,7 @@ class SimilarAsinTaskServiceRowKeyDedupeTest {
// 恢复后重试成功：只写一次，无重复记录
AtomicLong storedCounter = new AtomicLong(0);
stubSingleChunkMerge(chunk, rowsJson(List.of(row("r1", "1", "B0A0000001", "英国"))), storedCounter);
merge.invoke(service, task, null, null,
merge.invoke(service.pipelineSupport(), task, null, null,
List.of(row("r1", "1", "B0A0000001", "英国"), row("r1", "1", "B0A0000001", "英国")), Map.of());
assertEquals(1, storedCounter.get());
verify(transientPayloadStorageService, times(1)).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString());
backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceRowKeyIndexTest.java
+6 -5
View File
@@ -22,6 +22,7 @@ import org.apache.ibatis.builder.MapperBuilderAssistant;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import com.nanri.aiimage.modules.similarasin.service.support.SimilarAsinPipelineSupport;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.InjectMocks;
@@ -191,10 +192,10 @@ class SimilarAsinTaskServiceRowKeyIndexTest {
task.setId(7004L);
List<SimilarAsinResultRowDto> llmRows = List.of(row("r1", "1", "B0A0000001", "英国"), row("r3", "3", "B0A0000003", "美国"));
Method merge = SimilarAsinTaskService.class.getDeclaredMethod("mergeLlmRowsIntoChunk",
Method merge = SimilarAsinPipelineSupport.class.getDeclaredMethod("mergeLlmRowsIntoChunk",
FileTaskEntity.class, String.class, Integer.class, List.class, Map.class);
merge.setAccessible(true);
merge.invoke(service, task, null, null, llmRows, Map.of());
merge.invoke(service.pipelineSupport(), task, null, null, llmRows, Map.of());
verify(transientPayloadStorageService, times(6)).resolvePayload(anyString(), anyString());
verify(transientPayloadStorageService, times(2)).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString());
@@ -347,12 +348,12 @@ class SimilarAsinTaskServiceRowKeyIndexTest {
.thenThrow(new IllegalStateException("rustfs down"));
FileTaskEntity task = new FileTaskEntity();
task.setId(7004L);
Method merge = SimilarAsinTaskService.class.getDeclaredMethod("mergeLlmRowsIntoChunk",
Method merge = SimilarAsinPipelineSupport.class.getDeclaredMethod("mergeLlmRowsIntoChunk",
FileTaskEntity.class, String.class, Integer.class, List.class, Map.class);
merge.setAccessible(true);
BusinessException ex = assertThrows(BusinessException.class, () -> {
try {
merge.invoke(service, task, null, null, List.of(row("r1", "1", "B0A0000001", "英国")), Map.of());
merge.invoke(service.pipelineSupport(), task, null, null, List.of(row("r1", "1", "B0A0000001", "英国")), Map.of());
} catch (java.lang.reflect.InvocationTargetException e) {
throw e.getCause();
}
@@ -367,7 +368,7 @@ class SimilarAsinTaskServiceRowKeyIndexTest {
.thenReturn("stored:retry");
when(taskChunkMapper.selectOne(any())).thenReturn(chunks.get(0));
when(taskChunkMapper.update(any(), any())).thenReturn(1);
merge.invoke(service, task, null, null, List.of(row("r1", "1", "B0A0000001", "英国")), Map.of());
merge.invoke(service.pipelineSupport(), task, null, null, List.of(row("r1", "1", "B0A0000001", "英国")), Map.of());
verify(transientPayloadStorageService, times(1)).storeChunkPayloadVersioned(anyString(), any(), anyString(), any(), anyString());
}
}
backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskServiceTxBoundaryTest.java
+4 -3
View File
@@ -28,6 +28,7 @@ import org.apache.ibatis.builder.MapperBuilderAssistant;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import com.nanri.aiimage.modules.similarasin.service.support.SimilarAsinPipelineSupport;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.ArgumentCaptor;
@@ -169,7 +170,7 @@ class SimilarAsinTaskServiceTxBoundaryTest {
when(fileTaskMapper.selectById(TASK_ID)).thenReturn(task);
Object prepared = ReflectionTestUtils.invokeMethod(
service, "prepareSubmittedChunk", TASK_ID, request(false));
service.pipelineSupport(), "prepareSubmittedChunk", TASK_ID, request(false));
assertEquals(DigestUtil.sha256Hex(storedPayloadJson.get()),
ReflectionTestUtils.getField(prepared, "payloadHash"),
"prepare 阶段必须产出预计算的 payload 哈希（事务开始前可用）");
@@ -263,11 +264,11 @@ class SimilarAsinTaskServiceTxBoundaryTest {
@Test
void movedComputationMethodsCarryNoTransactionAnnotation() throws Exception {
assertNull(SimilarAsinTaskService.class
assertNull(SimilarAsinPipelineSupport.class
.getDeclaredMethod("prepareSubmittedChunk", Long.class, SimilarAsinSubmitResultRequest.class)
.getAnnotation(Transactional.class),
"prepareSubmittedChunk 不得带 @Transactional");
boolean persistUnannotated = java.util.Arrays.stream(SimilarAsinTaskService.class.getDeclaredMethods())
boolean persistUnannotated = java.util.Arrays.stream(SimilarAsinPipelineSupport.class.getDeclaredMethods())
.filter(method -> method.getName().equals("persistSubmittedChunk"))
.allMatch(method -> method.getAnnotation(Transactional.class) == null);
assertTrue(persistUnannotated, "persistSubmittedChunk 保持无注解（事务由调用方 inNewTransaction 控制）");