Spring Batch实现单文档异步处理与写入的方案咨询

针对你描述的场景（Reader一次性获取所有pending文档，每个文档独立执行依赖式处理和写入），以下是两种可行的实现方式：

方案一：使用AsyncItemProcessor + AsyncItemWriter #

该方案通过异步处理器将每个文档的处理逻辑提交到线程池，实现单文档独立线程执行，同时保证Processor内部的依赖式逻辑同步执行。

配置步骤 #

1. 定义线程池
   根据服务器配置设置合适的线程数，避免线程过多导致性能下降：

@Bean
public TaskExecutor batchTaskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(10); // 核心线程数
    executor.setMaxPoolSize(20); // 最大线程数
    executor.setQueueCapacity(50); // 任务队列容量
    executor.setThreadNamePrefix("batch-doc-processor-");
    executor.initialize();
    return executor;
}

2. 配置异步处理器
   将自定义的Processor包装为异步处理器，每个文档的处理会在独立线程中执行：

@Bean
public AsyncItemProcessor<EsDocument, EsDocument> asyncItemProcessor(CustomEsProcessor customProcessor, TaskExecutor batchTaskExecutor) {
    AsyncItemProcessor<EsDocument, EsDocument> asyncProcessor = new AsyncItemProcessor<>();
    asyncProcessor.setDelegate(customProcessor); // 注入你的自定义Processor
    asyncProcessor.setTaskExecutor(batchTaskExecutor);
    return asyncProcessor;
}

3. 配置异步写入器
   异步写入器会等待所有文档处理完成后，将结果提交给自定义Writer执行更新：

@Bean
public AsyncItemWriter<EsDocument> asyncItemWriter(CustomEsWriter customWriter) {
    AsyncItemWriter<EsDocument> asyncWriter = new AsyncItemWriter<>();
    asyncWriter.setDelegate(customWriter); // 注入你的自定义Writer
    return asyncWriter;
}

4. 配置Step
   结合异步处理器、写入器和线程池，chunk保持为1确保每个文档独立处理：

@Bean
public Step processPendingDocsStep(ItemReader<EsDocument> esItemReader,
                                  AsyncItemProcessor<EsDocument, EsDocument> asyncItemProcessor,
                                  AsyncItemWriter<EsDocument> asyncItemWriter) {
    return stepBuilderFactory.get("processPendingDocsStep")
            .<EsDocument, EsDocument>chunk(1)
            .reader(esItemReader)
            .processor(asyncItemProcessor)
            .writer(asyncItemWriter)
            .build();
}

5. 实现一次性读取的Reader
   自定义Reader从Elasticsearch批量获取所有pending文档，存入内存后通过ListItemReader提供数据：

@Bean
public ItemReader<EsDocument> esPendingItemReader(RestHighLevelClient esClient, ObjectMapper objectMapper) {
    SearchRequest searchRequest = new SearchRequest("your_document_index");
    SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
    sourceBuilder.query(QueryBuilders.termQuery("doc_status", "pending"));
    searchRequest.source(sourceBuilder);

    try {
        SearchResponse response = esClient.search(searchRequest, RequestOptions.DEFAULT);
        List<EsDocument> pendingDocs = Arrays.stream(response.getHits().getHits())
                .map(hit -> {
                    try {
                        return objectMapper.readValue(hit.getSourceAsString(), EsDocument.class);
                    } catch (IOException e) {
                        throw new RuntimeException("Failed to parse ES document", e);
                    }
                })
                .collect(Collectors.toList());
        return new ListItemReader<>(pendingDocs);
    } catch (IOException e) {
        throw new RuntimeException("Failed to fetch pending docs from ES", e);
    }
}

关键注意点 #

• 异常处理：可以重写AsyncItemWriter的write方法，捕获处理过程中的异常，比如标记失败文档状态：

@Override
public void write(List<? extends Future<EsDocument>> items) throws Exception {
    List<EsDocument> processedDocs = new ArrayList<>();
    for (Future<EsDocument> future : items) {
        try {
            processedDocs.add(future.get());
        } catch (ExecutionException e) {
            log.error("Document processing failed", e.getCause());
            // 可选：更新ES中文档状态为"failed"
        }
    }
    super.write(processedDocs);
}

• 事务隔离：每个文档的处理和写入会在独立事务中执行，避免单个文档失败影响其他文档。

方案二：配置多线程Step #

直接为Step配置线程池，让每个chunk（单文档）的读-处理-写流程在独立线程执行，适合更简单的场景。

配置步骤 #

1. 定义线程池（同方案一的batchTaskExecutor）
2. 实现一次性读取的Reader（同方案一的esPendingItemReader）
3. 配置多线程Step

@Bean
public Step multiThreadProcessStep(ItemReader<EsDocument> esPendingItemReader,
                                  CustomEsProcessor customProcessor,
                                  CustomEsWriter customWriter,
                                  TaskExecutor batchTaskExecutor) {
    return stepBuilderFactory.get("multiThreadProcessStep")
            .<EsDocument, EsDocument>chunk(1)
            .reader(esPendingItemReader)
            .processor(customProcessor)
            .writer(customWriter)
            .taskExecutor(batchTaskExecutor) // 为Step分配线程池
            .build();
}

关键注意点 #

• Reader线程安全：由于使用ListItemReader（基于内存列表的只读操作），天然线程安全，无需额外处理。
• 线程池大小：建议根据CPU核心数设置（比如核心数*2），避免线程上下文切换过多。

方案对比 #

• Async方案：更灵活，支持异步结果的批量处理和异常捕获，适合需要精细化控制处理流程的场景。
• 多线程Step方案：配置更简洁，直接实现chunk级别的多线程执行，适合简单的单文档独立处理场景。

内容的提问来源于stack exchange，提问作者SMalik