Spring Batch多线程RabbitMQ消费作业实现方案咨询

核心需求 #

实现Spring Batch作业从RabbitMQ队列消费消息并进行验证：

• 每条消息为JSON对象数组（示例格式：[{JSON object}, {JSON object}]）
• 提取JSON对象传递给处理器做验证
• 必须正确处理消息格式错误（如缺失括号、引号），手动执行作业时能返回错误消息

当前实现思路 #

• 采用基于chunking的容错Step，通过ThreadPoolTaskExecutor实现并行执行
• 自定义基于MultiResourceItemReader的多资源读取器，委托JsonItemReader读取JSON内容
• 为delegate的read()和open(new ExecutionContext())方法添加异常处理：读取发生格式错误类异常时，将当前资源对应的消息ID写入作业执行上下文，作业完成后可从中提取失败ID
• 使用SynchronizedItemStreamReader装饰自定义读取器，保证读取过程同步

遇到的问题 #

并发场景下异常记录存在缺失：

• 处理4条含格式错误的消息（每条含1000个对象）时，作业执行上下文仅记录3条失败消息ID，读取操作被判定为成功；缺失的那条消息甚至无错误日志输出
• 单独处理该缺失消息并搭配正确消息时，作业上下文能正常记录1条错误消息，说明单条/少量错误场景下异常处理有效，但多条错误消息时出现遗漏

核心疑问 #

当前方案整体是否可行？还是需要探索替代方案？核心需求是并行执行资源读取与JSON对象处理，同时准确捕获所有读取失败的消息ID。

当前Step代码 #

@Bean
public Step rabbitStep() {
    return new StepBuilder("rabbitValidateStep", jobRepository)
            .<ProcessRequest, Void>chunk(chunkSize, transactionManager)
            .reader(rabbitSynchronizedItemStreamReader())
            .processor(rabbitItemProcessor())
            .writer(item -> {
            })
            .faultTolerant()
            .skip(ParseException.class)
            .noRetry(ParseException.class)
            .noRollback(ParseException.class)
            .skip(ItemStreamException.class)
            .noRetry(ItemStreamException.class)
            .noRollback(ItemStreamException.class)
            .skipLimit(1000)
            .taskExecutor(rabbitDelegatingSecurityContextAsyncTaskExecutor(rabbitValidateThreadPoolTaskExecutor()))
            .throttleLimit(getWorkingThreads())
            .build();
}

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