Spring Batch轻量级批处理框架实战

public interface JobLauncher {

public JobExecution run(Job job, JobParameters jobParameters)
            throws JobExecutionAlreadyRunningException, JobRestartException,
                   JobInstanceAlreadyCompleteException, JobParametersInvalidException;
}

<!--Spring batch-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-batch</artifactId>
</dependency>
<!-- WEB依赖-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- lombok-->
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <version>1.18.20</version>
</dependency>
<!--  Mysql-->
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.47</version>
</dependency>
<!--  mybatis-->
<dependency>
    <groupId>com.baomidou</groupId>
    <artifactId>mybatis-plus-boot-starter</artifactId>
    <version>3.2.0</version>
</dependency>

server.port=9000
spring.datasource.url=jdbc:mysql://localhost:3306/test
spring.datasource.username=root
spring.datasource.passWord=12345
spring.datasource.driver-class-name=com.mysql.jdbc.Driver

CREATE TABLE `student` (
    `id` int(100) NOT NULL AUTO_INCREMENT,
    `name` varchar(45) DEFAULT NULL,
    `age` int(2) DEFAULT NULL,
    `address` varchar(45) DEFAULT NULL,
    PRIMARY KEY (`id`),
    UNIQUE KEY `id_UNIQUE` (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=203579 DEFAULT CHARSET=utf8 ROW_FORMAT=REDUNDANT

@Data
@Accessors(chain = true)
@NoArgsConstructor
@AllArgsConstructor
@ToString
@TableName("student")
public class Student {

    @TableId(value = "id", type = IdType.AUTO)
    private Long sId;

    @TableField("name")
    private String sName;

    @TableField("age")
    private Integer sAge;

    @TableField("address")
    private String sAddress;

}

@Mapper
@Repository
public interface StudentDao extends BaseMapper<Student> {
}

public class StudentVirtualDao {

    
    public List<Student> getStudents() {
        ArrayList<Student> students = new ArrayList<>();
        students.add(new Student(1L, "zs", 23, "Beijing"));
        students.add(new Student(2L, "ls", 23, "Beijing"));
        students.add(new Student(3L, "ww", 23, "Beijing"));
        students.add(new Student(4L, "zl", 23, "Beijing"));
        students.add(new Student(5L, "MQ", 23, "Beijing"));
        students.add(new Student(6L, "gb", 23, "Beijing"));
        students.add(new Student(7L, "lj", 23, "Beijing"));
        students.add(new Student(8L, "ss", 23, "Beijing"));
        students.add(new Student(9L, "zsdd", 23, "Beijing"));
        students.add(new Student(10L, "zss", 23, "Beijing"));
        return students;
    }
}

public interface StudentService {

    List<Student> selectStudentsFromDB();

    void insertStudent(Student student);
}

@Service
public class StudentServiceImpl implements StudentService {

    @Autowired
    private StudentDao studentDao;

    @Override
    public List<Student> selectStudentsFromDB() {
        return studentDao.selectList(null);
    }

    @Override
    public void insertStudent(Student student) {
        studentDao.insert(student);
    }
}

@Configuration
@EnableBatchProcessing
@SuppressWarnings("all")
public class BatchConfiguration {

    
    @Autowired
    public JobBuilderFactory jobBuilderFactory;

    
    @Autowired
    public StepBuilderFactory stepBuilderFactory;

    
    @Autowired
    public JobRepository jobRepository;

    
    @Autowired
    private JobLauncher jobLauncher;

    
    @Autowired
    private StudentService studentService;

    
    @Autowired
    private Job studentJob;

    
    @Bean
    public ItemWriter<Student> writer() {
        ItemWriter<Student> writer = new ItemWriter() {
            @Override
            public void write(List list) throws Exception {
                //debug发现是嵌套的List reader的线程List嵌套真正的List
                list.forEach((stu) -> {
                    for (Student student : (ArrayList<Student>) stu) {
                        studentService.insertStudent(student);
                    }
                });
            }
        };
        return writer;
    }

    
    @Bean
    public ItemReader<Student> reader() {
        ItemReader<Student> reader = new ItemReader() {
            @Override
            public Object read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
                //模拟数据获取
                StudentVirtualDao virtualDao = new StudentVirtualDao();
                return virtualDao.getStudents();
            }
        };
        return reader;
    }

    
    @Bean
    public ItemProcessor processor() {
        ItemProcessor processor = new ItemProcessor() {
            @Override
            public Object process(Object o) throws Exception {
                //debug发现o就是reader单次单线程读取的数据
                return o;
            }
        };
        return processor;
    }

    
    @Bean
    public Step studentStepOne() {
        return stepBuilderFactory.get("studentStepOne")
            .chunk(1)
            .reader(reader()) //加入reader
            .processor(processor())  //加入processor
            .writer(writer())//加入writer
            .build();
    }

    
    @Bean
    public Job studentJob() {
        return jobBuilderFactory.get("studentJob")
            .flow(studentStepOne())//加入step
            .end()
            .build();
    }


    
    @Scheduled(fixedRate = 5000)
    public void printMessage() {
        try {
            JobParameters jobParameters = new JobParametersBuilder()
                .addLong("time", System.currentTimeMillis())
                .toJobParameters();
            jobLauncher.run(studentJob, jobParameters);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

public class JobBuilderFactory {

   private JobRepository jobRepository;

   public JobBuilderFactory(JobRepository jobRepository) {
      this.jobRepository = jobRepository;
   }
   //返回JobBuilder
   public JobBuilder get(String name) {
      JobBuilder builder = new JobBuilder(name).repository(jobRepository);
      return builder;
   }
}

public class JobBuilder extends JobBuilderHelper<JobBuilder> {

   
   public JobBuilder(String name) {
      super(name);
   }

   
   public SimpleJobBuilder start(Step step) {
      return new SimpleJobBuilder(this).start(step);
   }

   
   public JobFlowBuilder start(Flow flow) {
      return new FlowJobBuilder(this).start(flow);
   }

   
   public JobFlowBuilder flow(Step step) {
      return new FlowJobBuilder(this).start(step);
   }
}

public class StepBuilder extends StepBuilderHelper<StepBuilder> {

   public StepBuilder(String name) {
      super(name);
   }

   
   public TaskletStepBuilder tasklet(Tasklet tasklet) {
      return new TaskletStepBuilder(this).tasklet(tasklet);
   }

   
   public <I, O> SimpleStepBuilder<I, O> chunk(int chunkSize) {
      return new SimpleStepBuilder<I, O>(this).chunk(chunkSize);
   }

   public <I, O> SimpleStepBuilder<I, O> chunk(CompletionPolicy completionPolicy) {
      return new SimpleStepBuilder<I, O>(this).chunk(completionPolicy);
   }

   public PartitionStepBuilder partitioner(String stepName, Partitioner partitioner) {
      return new PartitionStepBuilder(this).partitioner(stepName, partitioner);
   }

   public PartitionStepBuilder partitioner(Step step) {
      return new PartitionStepBuilder(this).step(step);
   }

   public JobStepBuilder job(Job job) {
      return new JobStepBuilder(this).job(job);
   }

   
   public FlowStepBuilder flow(Flow flow) {
      return new FlowStepBuilder(this).flow(flow);
   }
}