Spring Batch首次实现：Reader、Processor、Writer未执行求助

问题描述 #

我正在学习Spring Batch，首次实现后发现Reader、Processor和Writer中的代码并未执行，已验证CSV读取工具和数据库连接正常，以下是我的实现代码及控制台输出，寻求解决建议。

JustBatchApplication #

@SpringBootApplication
public class JustBatchApplication {

    public static void main(String[] args) {
        SpringApplication.run(JustBatchApplication.class, args);
    }

}

BatchConfiguration #

@Slf4j
@Configuration
@EnableBatchProcessing
@RequiredArgsConstructor
public class BatchConfiguration {
    
    private final EmployeeReader employeeReader;
    
    private final EmployeeProcessor employeeProcessor;
    
    private final EmployeeWriter employeeWriter;
    
    @Bean
    public Job employeeJob(JobRepository jobRepository, JobExecutionListener listener, Step step1) {
        return new JobBuilder("employeeJob", jobRepository)
                .incrementer(new RunIdIncrementer())
                .listener(listener)
                .flow(step1)
                .end()
                .build();
    }

    @Bean
    public Step step1(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
        return new StepBuilder("step1", jobRepository)
                .<Employee, Employee>chunk(10, transactionManager)
                .reader(employeeReader)
                .processor(employeeProcessor)
                .writer(employeeWriter)
                .build();
    }

}

EmployeeJobListener #

@Slf4j
@Component
@RequiredArgsConstructor
public class EmployeeJobListener implements JobExecutionListener {
    
    private final EmployeeReader employeeReader;
    
    @Override
    public void beforeJob(JobExecution jobExecution) {
        log.info("############## JOB STARTING ############## ");
    }

    @Override
    public void afterJob(JobExecution jobExecution) {
        if (jobExecution.getStatus() == BatchStatus.COMPLETED) {
            log.info("############## JOB COMPLETED ############## ");
            employeeReader.closeIterator();
        }
    }

}

EmployeeReader #

此处我使用已实现的CsvUtils将CSV的每一行映射到Employee类，已验证该工具工作正常，文件读取无问题。

@Slf4j
@Component
public class EmployeeReader implements ItemReader<Employee> {

    public EmployeeReader() {
        CsvUtils<Employee> csvEmployee = new CsvUtils<>(Employee.class);
        try {
            iterator = csvEmployee.iterator(new FileInfo(DirectoryPath.SRC_MAIN_RESOURCES.value, "employees"));
        } catch (IOException e) {
            throw new RuntimeException("IOException while reading employees file!", e);
        }
    }
    
    private MappingIterator<Employee> iterator;
    
    @Override
    public Employee read() {
                    
        if (iterator.hasNext()) {
            
            Employee employee = iterator.next();

            log.info("Reading item: {}", employee);
            
            return employee;
            
        } else {
            
            return null;
            
        }
        
    }
    
    public void closeIterator() {
        try {
            iterator.close();
        } catch (IOException e) {
            log.warn("IOException while trying to close the iterator!");
            e.printStackTrace();
        }
    }

}

EmployeeProcessor #

@Slf4j
@Component
public class EmployeeProcessor implements ItemProcessor<Employee, Employee> {

    @Override
    public Employee process(Employee employee) {
        log.info("Processing employee: {}", employee);
        return employee;
    }

}

EmployeeWriter #

@Slf4j
@Component
public class EmployeeWriter implements ItemWriter<Employee> {
    
    @Override
    public void write(Chunk<? extends Employee> chunk) throws Exception {
        for ( Employee employee : chunk ) {
            log.info("Writing to console : " + employee );
        }
    }
    
}

application.yaml #

spring:
  application:
    name: batch
    version: 1.0.0
  banner:
    location: classpath:banner.txt
  datasource:
    url: jdbc:postgresql://localhost:5432/BatchDB
    driverClassName: org.postgresql.Driver
    username: "userbatch"
    password: "passwordbatch"
  jpa:
    properties:
      hibernate:
        dialect: org.hibernate.dialect.PostgreSQLDialect
    database-platform: org.postgresql.Driver
    show-sql: 'true'
    hibernate:
      ddl-auto: update
  batch:
    job:
      enabled: true
server:
  port: '9999'

控制台输出 #

________       _____      ______  
___  __ )_____ __  /_________  /_ 
__  __  |  __ `/  __/  ___/_  __ \
_  /_/ // /_/ // /_ / /__ _  / / /
/_____/ \__,_/ \__/ \___/ /_/ /_/ 
                                  
 
Powered by Spring Boot 3.1.2

2023-09-17T01:09:40.442+02:00  INFO 44199 --- [           main] c.p.b.JustBatchApplication               : Starting JustBatchApplication using Java 17.0.5 with PID 44199 (/Users/xyz/Develop/Personal/projects/Training/batch/target/classes started by xyz in /Users/xyz/Develop/Personal/projects/Training/batch)
2023-09-17T01:09:40.443+02:00  INFO 44199 --- [           main] c.p.b.JustBatchApplication               : No active profile set, falling back to 1 default profile: "default"
2023-09-17T01:09:40.585+02:00  INFO 44199 --- [           main] o.s.b.c.c.a.BatchRegistrar               : Finished Spring Batch infrastructure beans configuration in 2 ms.
2023-09-17T01:09:40.632+02:00  INFO 44199 --- [           main] .s.d.r.c.RepositoryConfigurationDelegate : Multiple Spring Data modules found, entering strict repository configuration mode
2023-09-17T01:09:40.633+02:00  INFO 44199 --- [           main] .s.d.r.c.RepositoryConfigurationDelegate : Bootstrapping Spring Data JDBC repositories in DEFAULT mode.
2023-09-17T01:09:40.640+02:00  INFO 44199 --- [           main] .s.d.r.c.RepositoryConfigurationDelegate : Finished Spring Data repository scanning in 5 ms. Found 0 JDBC repository interfaces.
2023-09-17T01:09:40.647+02:00  INFO 44199 --- [           main] .s.d.r.c.RepositoryConfigurationDelegate : Multiple Spring Data modules found, entering strict repository configuration mode
2023-09-17T01:09:40.648+02:00  INFO 44199 --- [           main] .s.d.r.c.RepositoryConfigurationDelegate : Bootstrapping Spring Data JPA repositories in DEFAULT mode.
2023-09-17T01:09:40.650+02:00  INFO 44199 --- [           main] .s.d.r.c.RepositoryConfigurationDelegate : Finished Spring Data repository scanning in 1 ms. Found 0 JPA repository interfaces.
2023-09-17T01:09:40.772+02:00  INFO 44199 --- [           main] o.h.j.i.u.LogHelper                      : HHH000204: Processing PersistenceUnitInfo [name: default]
2023-09-17T01:09:40.794+02:00  INFO 44199 --- [           main] o.h.Version                              : HHH000412: Hibernate ORM core version 6.2.6.Final
2023-09-17T01:09:40.796+02:00  INFO 44199 --- [           main] o.h.c.Environment                        : HHH000406: Using bytecode reflection optimizer
2023-09-17T01:09:40.851+02:00  INFO 44199 --- [           main] o.h.b.i.BytecodeProviderInitiator        : HHH000021: Bytecode provider name : bytebuddy
2023-09-17T01:09:40.907+02:00  INFO 44199 --- [           main] o.s.o.j.p.SpringPersistenceUnitInfo      : No LoadTimeWeaver setup: ignoring JPA class transformer
[main] INFO com.zaxxer.hikari.HikariDataSource - HikariPool-1 - Starting...
[main] INFO com.zaxxer.hikari.pool.HikariPool - HikariPool-1 - Added connection org.postgresql.jdbc.PgConnection@204d9edf
[main] INFO com.zaxxer.hikari.HikariDataSource - HikariPool-1 - Start completed.
2023-09-17T01:09:41.124+02:00  INFO 44199 --- [           main] o.h.b.i.BytecodeProviderInitiator        : HHH000021: Bytecode provider name : bytebuddy
2023-09-17T01:09:41.361+02:00  INFO 44199 --- [           main] o.h.e.t.j.p.i.JtaPlatformInitiator       : HHH000490: Using JtaPlatform implementation: [org.hibernate.engine.transaction.jta.platform.internal.NoJtaPlatform]
2023-09-17T01:09:41.391+02:00  INFO 44199 --- [           main] j.LocalContainerEntityManagerFactoryBean : Initialized JPA EntityManagerFactory for persistence unit 'default'
2023-09-17T01:09:41.436+02:00  INFO 44199 --- [           main] o.s.b.c.r.s.JobRepositoryFactoryBean     : No database type set, using meta data indicating: POSTGRES
2023-09-17T01:09:41.453+02:00  INFO 44199 --- [           main] .c.a.BatchObservabilityBeanPostProcessor : No Micrometer observation registry found, defaulting to ObservationRegistry.NOOP
2023-09-17T01:09:41.457+02:00  INFO 44199 --- [           main] .c.a.BatchObservabilityBeanPostProcessor : No Micrometer observation registry found, defaulting to ObservationRegistry.NOOP
2023-09-17T01:09:41.459+02:00  INFO 44199 --- [           main] o.s.b.c.l.s.SimpleJobLauncher            : No TaskExecutor has been set, defaulting to synchronous executor.
2023-09-17T01:09:41.552+02:00  INFO 44199 --- [           main] c.p.b.JustBatchApplication               : Started JustBatchApplication in 1.286 seconds (process running for 1.637)
2023-09-17T01:09:41.554+02:00  INFO 44199 --- [ionShutdownHook] j.LocalContainerEntityManagerFactoryBean : Closing JPA EntityManagerFactory for persistence unit 'default'
[SpringApplicationShutdownHook] INFO com.zaxxer.hikari.HikariDataSource - HikariPool-1 - Shutdown initiated...
[SpringApplicationShutdownHook] INFO com.zaxxer.hikari.HikariDataSource - HikariPool-1 - Shutdown completed.

已通过保存Employee验证PostgreSQL连接正常。

解决建议 #

• 强制作业重复执行：Spring Batch默认会跳过已执行成功的作业（通过元数据表记录），可以通过以下方式强制每次启动都执行：

  1. 修改application.yaml添加配置：

     spring:
       batch:
         job:
           enabled: true
         jdbc:
           initialize-schema: always
     

  2. 启动时通过命令行参数指定作业：java -jar your-app.jar --spring.batch.job.names=employeeJob

• 确认作业监听器注入正确性：在BatchConfiguration的employeeJob方法中，显式指定监听器Bean的名称，避免注入歧义：

  @Bean
  public Job employeeJob(JobRepository jobRepository, @Qualifier("employeeJobListener") JobExecutionListener listener, Step step1) {
      return new JobBuilder("employeeJob", jobRepository)
              .incrementer(new RunIdIncrementer())
              .listener(listener)
              .flow(step1)
              .end()
              .build();
  }
  

• 排查Reader数据加载情况：在EmployeeReader的构造函数中添加日志，确认CSV文件是否成功加载且包含数据：

  public EmployeeReader() {
      CsvUtils<Employee> csvEmployee = new CsvUtils<>(Employee.class);
      try {
          iterator = csvEmployee.iterator(new FileInfo(DirectoryPath.SRC_MAIN_RESOURCES.value, "employees"));
          log.info("CSV文件加载完成，是否有数据：{}", iterator.hasNext());
      } catch (IOException e) {
          throw new RuntimeException("IOException while reading employees file!", e);
      }
  }
  

• 显式配置事务管理器：确保Spring Batch能获取到正确的事务管理器，添加以下Bean到配置类：

  @Bean
  public PlatformTransactionManager transactionManager(EntityManagerFactory entityManagerFactory) {
      return new JpaTransactionManager(entityManagerFactory);
  }
  

• 检查Spring Batch元数据表：查看PostgreSQL中的BATCH_JOB_EXECUTION、BATCH_STEP_EXECUTION表，确认作业是否有执行记录，以及执行状态和异常信息，这能直接定位作业未执行的原因。

内容的提问来源于stack exchange，提问作者Paul Marcelin Bejan