概述

快速上手, 以最快的方式和配置運(yùn)行和了解Spring Batch批處理; 通過(guò)bean注入的方式, 可以使用spring-boot-starter-batch進(jìn)行相關(guān)bean的管理和配置

但是這種方式不夠靈活, 因?yàn)轫?xiàng)目不一定在spring的環(huán)境中執(zhí)行丈钙，所以需要另一種方式灶平。結(jié)合源碼不難發(fā)現(xiàn)Batch執(zhí)行Job的核心代碼和創(chuàng)建過(guò)程

非Bean注入方式

通過(guò)人工進(jìn)行對(duì)象的創(chuàng)建

Maven

<dependency>
            <groupId>org.springframework.batch</groupId>
            <artifactId>spring-batch-core</artifactId>
            <version>4.3.1</version>
 </dependency>

Code

public class RunDemo {

    static class MyBatchConfigurer extends DefaultBatchConfigurer {
        @Override
        public void setDataSource(DataSource dataSource) {
        }
    }

    public static void main(String[] args) throws Exception {
        File dataFile = new File("data.json");
        if (!dataFile.exists()) {
            dataFile.createNewFile();
        }
        FileOutputStream os = new FileOutputStream(dataFile);
        os.write("[{\"city\":\"beijing\",\"name\":\"jaychou\"},".getBytes());
        os.write("{\"city\":\"beijing\",\"name\":\"jaychou2\"}]".getBytes());
        os.flush();
        os.close();

        MyBatchConfigurer context = new MyBatchConfigurer();
        context.initialize();

        JsonItemReader<Map> reader = new JsonItemReaderBuilder<Map>().name("reader")
                .resource(new FileSystemResource(dataFile))
                .jsonObjectReader(new JacksonJsonObjectReader<>(Map.class))
                .saveState(true).build();

        FlatFileItemWriter<String> writer = new FlatFileItemWriterBuilder<String>().name("writer")
                .resource(new FileSystemResource("target/output.txt")).lineAggregator(new PassThroughLineAggregator<>()).build();

        Step step = new StepBuilderFactory(context.getJobRepository(), context.getTransactionManager())
                .get("step1").<Map, String>chunk(10).reader(reader).writer(writer)
                .processor((ItemProcessor<Map, String>) (v) -> "__" + v.toString())
                .build();

        Job job = new JobBuilderFactory(context.getJobRepository()).get("job").start(step).build();
        context.getJobLauncher().run(job, new JobParameters());
    }
}

Bean注入方式

Maven

 <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-batch</artifactId>
      <version>2.4.1</version>
  </dependency>

Code

SpringBatchApplication

@SpringBootApplication(exclude = {DataSourceAutoConfiguration.class})
public class SpringBatchApplication {
  public static void main(String[] args) {
    SpringApplication.run(SpringBatchApplication.class, args);
  }
}

RunDemo

@Configuration
@EnableBatchProcessing
public class RunDemo {
    @Autowired
    private JobBuilderFactory jobBuilders;
    @Autowired
    private StepBuilderFactory stepBuilders;

    @Component
    static class MyBatchConfigurer extends DefaultBatchConfigurer {
        @Override
        public void setDataSource(DataSource dataSource) {
        }
    }
    @Bean
    public Job demoJob() throws Exception {
        File dataFile = new File("data.json");
        if (!dataFile.exists()) {
            dataFile.createNewFile();
        }
        FileOutputStream os = new FileOutputStream(dataFile);
        os.write("[{\"city\":\"beijing\",\"name\":\"jaychou\"},".getBytes());
        os.write("{\"city\":\"beijing\",\"name\":\"jaychou2\"}]".getBytes());
        os.flush();
        os.close();

        JsonItemReader<Map> reader = new JsonItemReaderBuilder<Map>().name("reader")
                .resource(new FileSystemResource(dataFile))
                .jsonObjectReader(new JacksonJsonObjectReader<>(Map.class))
                .saveState(true).build();

        FlatFileItemWriter<String> writer = new FlatFileItemWriterBuilder<String>().name("writer")
                .resource(new FileSystemResource("target/output.txt")).lineAggregator(new PassThroughLineAggregator<>()).build();

        Step step = stepBuilders.get("step1").<Map, String>chunk(10).reader(reader).writer(writer)
                .processor((ItemProcessor<Map, String>) (v) -> "__" + v.toString())
                .build();

        return jobBuilders.get("job").start(step).build();
    }
}

補(bǔ)充

如果想要其他的輸入和輸出, 如KafkaItemReader,KafkaItemWriter, 則需要引入如下的依賴(lài)包

    <dependency>
      <groupId>org.springframework.batch</groupId>
      <artifactId>spring-batch-infrastructure</artifactId>
      <version>4.3.1</version>
    </dependency>

   <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka_2.12</artifactId>
      <version>2.0.1</version>
    </dependency>

demo運(yùn)行結(jié)果, 會(huì)生成output.txt

__{city=beijing, name=jaychou}
__{city=beijing, name=jaychou2}