spring - 使用分区器 Spring Batch 限制线程

Question

我的 Spring 批处理应用程序消耗了太多资源（+4 go Ram）。

当我查看 jvm 时，应用程序创建了 10 个线程。

• 我使用分区程序在没有调度程序的情况下逐个文件处理文件

• jobExecutionListener 用于在执行结束时停止批处理

  @Bean
  public Job mainJob() throws IOException {
  SimpleJobBuilder mainJob = this.jobBuilderFactory.get("mainJob")
              .start(previousStep())    
              .next(partitionStep())
              .next(finalStep())
              .listener(jobExecutionListener(taskExecutor()));;
      return mainJob.build();
  }
  
  @Bean
  public Step partitionStep() throws IOException {
      Step mainStep = stepBuilderFactory.get("mainStep")
              .<InOut, InOut>chunk(1)
              .reader(ResourceReader())
              .processor(processor())
              .writer(itemWriter())
              .build();
  
      return this.stepBuilderFactory.get("partitionStep")
              .partitioner(mainStep)
              .partitioner("mainStep", partitioner())
              .build();
  }
  
  @Bean(name = "taskExecutor")
  public ThreadPoolTaskExecutor taskExecutor() {
      ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
      taskExecutor.setCorePoolSize(1);
      taskExecutor.setMaxPoolSize(1);
      taskExecutor.setQueueCapacity(1);
      taskExecutor.setThreadNamePrefix("MyBatch-");
      taskExecutor.initialize();
  
      return taskExecutor;
  }
  
  //This jobExecutionListener stop the batch
  @Bean
  public JobExecutionListener jobExecutionListener(@Qualifier("taskExecutor") 
  ThreadPoolTaskExecutor executor) {
      return new JobExecutionListener() {
          private ThreadPoolTaskExecutor taskExecutor = executor;
          @Override
          public void beforeJob(JobExecution jobExecution) {
          }
  
          @Override
          public void afterJob(JobExecution jobExecution) {
              taskExecutor.shutdown();
              System.exit(0);
          }
      };
  }
  
  @Bean
  public Partitioner partitioner() {
      MultiResourcePartitioner partitioner = new MultiResourcePartitioner();
      ResourcePatternResolver patternResolver = new 
      PathMatchingResourcePatternResolver();
  
      try {
          partitioner.setResources(patternResolver.getResources(FILE + 
      configProperties.getIn()+ "/*.xml"));
      } catch (IOException e) {
          throw new RuntimeException("I/O problems when resolving the input file pattern.",e);
      }
      partitioner.setKeyName("file");
      return partitioner;
  }
  

如何在单线程中应用我的应用程序？任务执行器不起作用。

标签： springspring-batch

Answer 1

您的应用创建了 10 个线程，但这些线程不一定是 Spring Batch 线程。根据您的配置，MyBatch-应该只创建一个带前缀的线程。

此外，您将任务执行器声明为 bean，但您没有在分区步骤上设置它。你partitionStep应该是这样的：

@Bean
public Step partitionStep() throws IOException {
   Step mainStep = stepBuilderFactory.get("mainStep")
        .<InOut, InOut>chunk(1)
        .reader(ResourceReader())
        .processor(processor())
        .writer(itemWriter())
        .build();

   return this.stepBuilderFactory.get("partitionStep")
        .step(mainStep) // instead of .partitioner(mainStep)
        .partitioner("mainStep", partitioner())
        .taskExecutor(taskExecutor())
        .build();
}

如何在单线程中应用我的应用程序？任务执行器不起作用。

在分区步骤上设置任务执行器后，您应该看到此步骤由您的ThreadPoolTaskExecutor. 但是，我没有看到将单个线程用于分区步骤的好处，因为这种设置的通常目标是并行处理分区（本地使用多个线程或远程使用多个工作 JVM）。

附带说明一下，最好使用 中的 Job 侦听器关闭任务执行器afterJob，但不要System.exit. 您需要让 JVM 正常关闭。

希望这可以帮助。