Spring Batch 未执行 StepExecutionListener beforeStep 或 afterStep 方法答案

【问题标题】：Spring Batch not executing StepExecutionListener beforeStep or afterStep methodsSpring Batch 未执行 StepExecutionListener beforeStep 或 afterStep 方法
【发布时间】：2021-07-01 22:41:00
【问题描述】：

更新：我正在重新运行我的代码，实际上发现我的所有侦听器都没有真正工作......

我有一个 Spring Batch 应用程序，我正在覆盖 StepExecutionListener 并提供我自己的实现。我正在使用 TaskletStep 注册它，但是，我从来没有看到 beforeStep/afterStep 方法应该输出的日志消息：

MyStepExecutionListener.java

public class MyStepExecutionListener implements StepExecutionListener {

 @Override
    public void beforeStep(StepExecution stepExecution) {
        // begin my own custom implementation
        LOGGER.info("Before the step!");
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
         // begin my own custom implementation
         LOGGER.info("After the step!");
        return stepExecution.getStatus();

    }
}

我在 BatchConfig.java 类中将我的 Tasklet 步骤定义为：

public BatchConfig {

@Bean
    public static org.springframework.batch.core.scope.JobScope jobScope() {
        org.springframework.batch.core.scope.JobScope jobScope = new org.springframework.batch.core.scope.JobScope();
        jobScope.setProxyTargetClass(true);
        jobScope.setAutoProxy(true);
        return jobScope;
    }

    @Bean
    public static org.springframework.batch.core.scope.StepScope stepScope() {
        org.springframework.batch.core.scope.StepScope stepScope = new org.springframework.batch.core.scope.StepScope();
        stepScope.setProxyTargetClass(true);
        stepScope.setAutoProxy(true);
        return stepScope;
    }

    @Bean
//    @StepScope
    public StepExecutionListener stepExecutionListener() {
        return new MyStepExecutionListener();
    }

 @Bean
    @Qualifier("s3FlatfFileReaderForMktgOffrs")
    @StepScope
    public S3FlatFileItemReader<FieldSet> s3FlatfFileReaderForMktgOffrs() {
        return new S3FlatFileItemReader<>(lineMapper());
    }

 @Bean
    @Qualifier("s3FlatfFileReaderCustom")
    @StepScope
    public S3FlatFileItemReader<FieldSet> s3FlatfFileReaderCustom() {
        // Custom class that Extends FlatFileItemReader
        return new S3FlatFileItemReader<>(lineMapper());
    }

  @Bean
    @Qualifier("myCustomFileItemReader")
    @StepScope
    public ItemStreamReader<List<FieldSet>> myCustomFileItemReader(
            @Value("#{jobParameters}") Map jobParameters) {

        String fileName = (String) jobParameters.get("fileName");
        String region = (String) jobParameters.get("region");
        String bucketName = awsS3EastBucket;
        if (StringUtils.equals(region, Regions.US_WEST_2.getName())) {
            bucketName = awsS3WestBucket;
        }

        // Custom class that Extends FlatFileItemReader
        S3FlatFileItemReader<FieldSet> s3FileItemReader = s3FlatfFileReaderCustom();

     
         s3FileItemReader.setResource(S3_PROTOCOL_PREFIX + bucketName + SLASH + fileName);
        }

        s3FileItemReader.setStrict(false);
        s3FileItemReader.setLinesToSkip(1);
        s3FileItemReader.setSaveState(false);

        AggregateItemReader aggregateItemReader = new AggregateItemReader(s3FileItemReader) {
            @Override
            protected String getItemKey(FieldSet item) {
                return item.readString(FIRST_NAME) + "-" +
                        item.readString(LAST_NAME);
            }
        };
        SynchronizedItemStreamReader<List<FieldSet>> fieldSetSynchronizedItemStreamReader = new SynchronizedItemStreamReader<>();
        fieldSetSynchronizedItemStreamReader.setDelegate(aggregateItemReader);
        return fieldSetSynchronizedItemStreamReader;
    }



@Bean(name = "myCustomStep")
    @Scope("prototype")
    @SuppressWarnings("unchecked")
    public Step myCustomStep(PlatformTransactionManager transactionManager) {
        TaskletStep step = stepBuilderFactory.get("myCustomStep")
                .<List<FieldSet>, List<MyPayLoadRecord>>chunk(250)
                .reader(myCustomFileItemReader(OVERRIDDEN_BY_EXPRESSION))
                .processor(myCustomProcessor())
                .writer(myCustomWriter())
                .faultTolerant()
                .skipPolicy(new AlwaysSkipItemSkipPolicy())
                .skip(DataValidationException.class)
                .listener(stepExecutionListener())
                .listener(new CustomReaderListener())
                .listener(new CustomProcessListener())
                .listener(new CustomWriteListener())
                .listener(new CustomSkipListener())
                .taskExecutor(batchTaskExecutor())
                .throttleLimit(maxThreads)
                .build();
        step.setTransactionManager(transactionManager);
        //step.registerStepExecutionListener(stepExecutionListener());
        step.registerChunkListener(new CustomChunkListener());
        return step;
    }
}

我已经注释掉了step.registerStepExecutionListener(stepExecutionListener()); 并尝试设置监听器，如上所示，但两种实现都不起作用。我的印象是我应该只实现 StepExecutionListener 然后用 TaskletStep 注册它 - 我在这里遗漏了什么吗？

【问题讨论】：

stepExecutionListener() 是如何定义的？
@RoddyoftheFrozenPeas 我现在在上面添加了 Bean 定义！
你如何验证监听器“工作”？调试会话中的断点，或者只是检查日志？（如果是第二个，您是否确认您的日志记录级别配置正确？最简单的方法是在 MyStepExecutionListener 的构造函数中放置一个日志记录语句）
你的afterStep 方法也缺少它的返回语句，所以它甚至不应该被编译。
一个步骤应该有@JobScope，我不明白你为什么要使用@Scope("prototype")？

标签： java spring spring-batch

【解决方案1】：

我不明白将您的步骤设置为原型范围 bean 的原因，但我无法使用类似于您共享的设置（使用 Spring Batch v4.3.3）重现该问题。这是一个简单的例子：

import java.util.Arrays;

import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.item.support.ListItemReader;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;

@Configuration
@EnableBatchProcessing
public class SO68217675 {

    @Bean
//    @StepScope // works with a step-scoped bean as well
    public StepExecutionListener stepExecutionListener() {
        return new MyStepExecutionListener();
    }

    @Bean
//    @Scope("prototype") // works with a prototype bean as well, but I don't understand the reason for that scope
    public Step step(StepBuilderFactory stepBuilderFactory) {
        return stepBuilderFactory.get("step")
                .<Integer, Integer>chunk(2)
                .reader(new ListItemReader<>(Arrays.asList(1, 2, 3, 4)))
                .writer(items -> items.forEach(System.out::println))
                .listener(stepExecutionListener())
                .build();
    }

    @Bean
    public Job job(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory) {
        return jobBuilderFactory.get("job")
                .start(step(stepBuilderFactory))
                .build();
    }

    static class MyStepExecutionListener implements StepExecutionListener {

        @Override
        public void beforeStep(StepExecution stepExecution) {
            System.out.println("MyStepExecutionListener.beforeStep");
        }

        @Override
        public ExitStatus afterStep(StepExecution stepExecution) {
            System.out.println("MyStepExecutionListener.afterStep");
            return ExitStatus.COMPLETED;
        }
    }

    public static void main(String[] args) throws Exception {
        ApplicationContext context = new AnnotationConfigApplicationContext(SO68217675.class);
        JobLauncher jobLauncher = context.getBean(JobLauncher.class);
        Job job = context.getBean(Job.class);
        jobLauncher.run(job, new JobParameters());
    }

}

此示例打印以下内容（如预期的那样）：

MyStepExecutionListener.beforeStep
1
2
3
4
MyStepExecutionListener.afterStep

【讨论】：

我添加了我的 stepScope 和 jobScope 的方法实现，当我的应用程序启动时我的 Reader 方法有一个奇怪的输出：org.springframework.batch.item.ItemStreamReader 是一个接口。不会查询实现类以获取基于注释的侦听器配置。如果在 Bean 方法上使用 StepScope，请务必返回实现类，以便可以使用监听器注解。
我实际上在我的阅读器方法中返回了一个 SynchronizedItemStreamReader> 所以我将方法签名从 ItemStreamReader 更改为那个，它似乎删除了错误消息，但我仍然没有查看记录的步骤。我觉得这可能与阅读器有关，或者可能与正在使用的所有代理/步骤/作业范围有关。
到目前为止，我能让任何 Listener 工作的唯一方法是直接在“myCustomProcessor()”之类的类上实现“StepExecutionListener”。当我在单独的类中定义这些侦听器并尝试使用TaskletStep注册它们时，我的配置定义方式似乎无法找到/调用Spring批处理框架，即“beforeStep” /afterStep"
这太奇怪了——我发现了问题。我重新阅读了您的答案（再次感谢您）并在我的 BatchConfiguration.java 类中添加了监听器，批处理作业的所有 Bean 都在该类中。我使用了一个静态嵌套类，就像你为 StepListener 所做的那样，你猜怎么着？有用！我仍然有点困惑，为什么返回在此 BatchConfiguration 之外定义的 StepListener 的自定义类不起作用……框架找不到它吗？我认为将 StepListener 定义为 Bean 会将其添加到 IoC
哎呀，我现在看得更清楚了，听众一直在工作。我认为他们不是，因为我在之前/之后的方法中有记录器，但他们从未记录过。但是我扔了一个 System.outprint.ln 并且我看到它在工作，所以问题出在 LOGGER 上。使用静态嵌套类帮助我找出答案。仍然不确定为什么会这样，但是哦，好吧

【解决方案2】：

根据@MahmoudBenHassine 的回答 - 通过将侦听器（Step 或其中任何一个）实现为静态嵌套类，我能够在 BatchConfiguration.java 类中使用 Slf4j LOGGER，并看到确实是侦听器工作中。不过，当我在自己的包/类中定义侦听器时，为什么这似乎不起作用，我仍然感到困惑。然后我开始怀疑记录器，基本上，这就是我验证它是否“工作”的方式。因此，我在我的 StepExecutionerListener 中放置了一些 System.out.println("") 虚拟输出（我遇到的问题是在它自己的自定义类/包中定义的）和 lo-and-behold ，我看到了输出。

因此，Listeners 一直在工作，但是 LOGGERS 没有工作，而且我的验证方法很差，因为它隐藏/隐藏了主要问题（使用 LOGGERS 作为 Step/Chunk/Job Listener 是否工作的证据）。

根据这篇文章，它让我更加好奇，我的 POM 中的 slf4j/log4j 依赖项出了点问题，或者我遗漏了一些东西：

Spring Batch produces no logging output when launched from the command line

此外，我按照本文设置了我的 POM.xml 以记录日志，该文章建议从所有 spring-boot-starters 中删除所有 spring-boot-starter-logging（我这样做了），并添加 1 个 spring-boot-starter-log4j2 依赖项，如果你打算使用 log4j（就像我一样）：

https://www.baeldung.com/spring-boot-logging

无论如何，我在我的 POM 中定义的spring-boot-starter-batch 依赖项中删除了spring-boot-starter-logging 的<exclusion>，并且日志现在（似乎）在自定义类监听器（Step/Chunk/Job）中工作正常.

当我在 IntelliJ 中检查我的项目依赖项时，我有：

log4j:log4j:1.2.15
org.apache.logging.log4j:log4j-api:2.13.3
org.apache.logging.log4j:log4j-core:2.13.3
org.apache.logging.log4j:log4j-jul:2.13.3
org.apache.logging.log4j:log4j-slf4j-impl:2.13.3
org.slf4j:jcl-over-slf4j:1.7.30
org.slf4j:jul-to-slf4j:1.7.30
org.slf4j:log4j-over-slf4j:1.7.30
org.slf4j:slf4j-api:1.7.30

我注意到通过删除 spring-boot-starter-batch 上的排除项，如上所述，我看到添加了此依赖项：

org.apache.logging.log4j:log4j-to-slf4j:2.13.3

所以我假设 Spring Batch 在触发侦听器before/after LOGGING 方法时需要这个。希望这对任何人都有帮助！

【讨论】：