Spring Batch任务step1未执行完成step2提前触发问题求助

问题根本原因 #

1. SNS发送逻辑执行时机完全错误：你在PublishSnsTopic类中把读取S3文件、发送SNS的逻辑写在了@PostConstruct注解的方法里，这个方法会在Spring容器初始化当前Bean的时候就执行，远早于Job启动、step1执行的时间。而Tasklet的execute方法才是step2运行时真正要执行的逻辑，你现在的execute方法是空实现，直接返回完成，所以才会出现step1还没跑，SNS已经发完了的现象。
2. 输出文件关闭逻辑未生效：BadStudentWriter里的@AfterStep回调方法名拼写错误（你写的是aferStep，少了字母t，正确应为afterStep），导致Spring不会触发这个回调，csvPrinter和S3输出流没有正常关闭，就算step1执行完，S3上的文件可能还没完成写入提交，也会导致后续读取异常。

修复方案 #

1. 修复BadStudentWriter的回调方法名 #

把拼写错误的aferStep改为afterStep，确保文件写完后流正常关闭，S3文件完整提交：

@AfterStep
void afterStep() {
    this.csvPrinter.close()
    this.writer.close() // 额外把PrintStream也关闭，确保资源释放
}

2. 重写PublishSnsTopic逻辑 #

移除@PostConstruct注解，把读取文件、发送SNS的逻辑全部移到execute方法中，同时去掉类上多余的@Configuration注解：

@Service
class PublishSnsTopic implements Tasklet {

    @Autowired
    ResourceLoader resourceLoader

    @Autowired
    FileProperties fileProperties

    void publishTopic(SnsClient snsClient, String message, String arn) {
        try {
            PublishRequest request = PublishRequest.builder()
                                        .message(message)
                                        .topicArn(arn)
                                        .build()
            snsClient.publish(request)
        } catch (SnsException e) {
          log.error "SOMETHING WENT WRONG", e
        }
    }

    @Override
    RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
        String badStudentCSVFileName = "s3://students/failedStudents.csv"
        // 先判断文件是否存在，避免异常
        WritableResource resource = this.resourceLoader.getResource(badStudentCSVFileName)
        if (!resource.exists()) {
            log.error "failedStudents.csv not found in S3"
            throw new FileNotFoundException("failedStudents.csv not found")
        }
        // 读取文件内容
        try(Reader badStudentReader = new InputStreamReader(resource.inputStream)) {
            List<BadStudent> badStudents = new CsvToBeanBuilder(badStudentReader)
                    .withSeparator((char)'|')
                    .withType(BadStudent)
                    .withFieldAsNull(CSVReaderNullFieldIndicator.BOTH)
                    .build()
                    .parse()
            String messageBody = badStudents.collect {it.id}.join(",")
            // 发送SNS
            try(SnsClient snsClient = SnsClient.builder().build()) {
                publishTopic(snsClient, messageBody, this.fileProperties.topicArn)
            }
        }
        return RepeatStatus.FINISHED
    }
}

额外优化建议 #

• 可以在step1和step2之间加个小的校验步骤，确认failedStudents.csv存在且大小正常，再执行step2，避免极端情况下S3最终一致性导致的读取失败。
• 如果数据量不大，也可以直接把不及格学生列表存在Job Execution Context中，step2直接从上下文拿数据，不需要再读一遍S3，效率更高也更可靠。

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