如何使用Spring Batch在并行模式下运行一个步骤

Question

我正在做Spring批次。我有一个分区步骤（对象列表），然后是读者和作家的从属步骤。

我想以并行模式执行processStep。因此，我希望每个分区都有一个特定的读写器实例。

目前，创建的分区使用相同的读写器实例。因此，这些操作是在串行模式下完成的：读取和写入第一个分区，然后在第一个分区完成后对下一个分区执行相同的操作。

Spring Boot配置类：

@Configuration
@Import({ DataSourceConfiguration.class})
public class BatchConfiguration {

    private final static int COMMIT_INTERVAL = 1;

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

   @Autowired
   @Qualifier(value="mySqlDataSource")
   private DataSource mySqlDataSource;

   public static int GRID_SIZE = 3;

   public static List<Pojo> myList;

   @Bean
   public Job myJob() throws UnexpectedInputException, ParseException, NonTransientResourceException, Exception {

      return jobBuilderFactory.get("myJob")
            .incrementer(new RunIdIncrementer())
            .start(partitioningStep())
            .build();
  }

  @Bean(name="partitionner")
  public MyPartitionner partitioner() {

    return new MyPartitionner();
  }

  @Bean
  public SimpleAsyncTaskExecutor taskExecutor() {

    SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
    taskExecutor.setConcurrencyLimit(GRID_SIZE);
    return taskExecutor;
  }

  @Bean
  public Step partitioningStep() throws NonTransientResourceException, Exception {

    return stepBuilderFactory.get("partitioningStep")
              .partitioner("processStep", partitioner())
              .step(processStep())
              .taskExecutor(taskExecutor())
              .build();
  }

  @Bean
  public Step processStep() throws UnexpectedInputException, ParseException, NonTransientResourceException, Exception {

    return stepBuilderFactory.get("processStep")
            .<List<Pojo>, List<Pojo>> chunk(COMMIT_INTERVAL)
            .reader(processReader())
            .writer(processWriter())
            .taskExecutor(taskExecutor())
            .build();
  }

  @Bean
  public ProcessReader processReader() throws UnexpectedInputException, ParseException, NonTransientResourceException, Exception {

    return new ProcessReader();
  }

  @Bean
  public ProcessWriter processWriter() {

    return new ProcessWriter();
  }
}

partitionner类

public class MyPartitionner implements Partitioner{

@Autowired
private IService service;

@Override
public Map<String, ExecutionContext> partition(int gridSize) {

    // list of 300 object partitionned like bellow
    ...
    Map<String, ExecutionContext> partitionData = new HashMap<String, ExecutionContext>();

    ExecutionContext executionContext0 = new ExecutionContext();
    executionContext0.putString("from", Integer.toString(0));
    executionContext0.putString("to", Integer.toString(100));
    partitionData.put("Partition0", executionContext0);

    ExecutionContext executionContext1 = new ExecutionContext();
    executionContext1.putString("from", Integer.toString(101));
    executionContext1.putString("to", Integer.toString(200));
    partitionData.put("Partition1", executionContext1);

    ExecutionContext executionContext2 = new ExecutionContext();
    executionContext2.putString("from", Integer.toString(201));
    executionContext2.putString("to", Integer.toString(299));
    partitionData.put("Partition2", executionContext2);

    return partitionData;
 }
}

读者类

    public class ProcessReader implements ItemReader<List<Pojo>>, ChunkListener {

    @Autowired
    private IService service;

    private StepExecution stepExecution;

    private static List<String> processedIntervals = new ArrayList<String>();

    @Override
    public List<Pojo> read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {

        System.out.println("Instance reference: "+this.toString());

        if(stepExecution.getExecutionContext().containsKey("from") && stepExecution.getExecutionContext().containsKey("to")){

            Integer from = Integer.valueOf(stepExecution.getExecutionContext().get("from").toString());
            Integer to = Integer.valueOf(stepExecution.getExecutionContext().get("to").toString());

            if(from != null && to != null && !processedIntervals.contains(from + "" + to) && to < BatchConfiguration.myList.size()){
                processedIntervals.add(String.valueOf(from + "" + to));
                return BatchConfiguration.myList.subList(from, to);
            }
        }

        return null;
    }

    @Override
    public void beforeChunk(ChunkContext context) {

        this.stepExecution = context.getStepContext().getStepExecution();
    }

    @Override
    public void afterChunk(ChunkContext context) { }

    @Override
    public void afterChunkError(ChunkContext context) { }

    }
  }

作家班

 public class ProcessWriter implements ItemWriter<List<Pojo>>{

    private final static Logger LOGGER = LoggerFactory.getLogger(ProcessWriter.class);

    @Autowired
    private IService service;

    @Override
    public void write(List<? extends List<Pojo>> pojos) throws Exception {

        if(!pojos.isEmpty()){
            for(Pojo item : pojos.get(0)){
                try {
                    service.remove(item.getId());
                } catch (Exception e) {
                    LOGGER.error("Error occured while removing the item [" + item.getId() + "]", e);
                }
            }
        }
    }
 }

你能告诉我我的密码有什么问题吗？

Answer 1

通过将@StepScope添加到我的读写器bean声明中解决：

@Configuration
@Import({ DataSourceConfiguration.class})
public class BatchConfiguration {

   ...

   @Bean
   @StepScope
    public ProcessReader processReader() throws UnexpectedInputException, ParseException, NonTransientResourceException, Exception {

      return new ProcessReader();
   }

   @Bean
   @StepScope
   public ProcessWriter processWriter() {

     return new ProcessWriter();
   }

   ...

}

通过这种方式，您可以看到每个分区都有一个不同的chunck（读写器）实例。