Spring Batch輕量級批處理框架實戰

Posted on 2021-10-18 by WalkonNet

1 實戰前的理論基礎

1.1 Spring Batch是什麼

Spring Batch 是一個輕量級、全面的批處理框架，旨在支持開發對企業系統日常運營至關重要的強大的批處理應用程序。同時使開發人員在必要時可以輕松訪問和利用更先進的企業服務。Spring Batch 不是調度框架，它旨在與調度程序一起工作，而不是取代調度程序。

1.2 Spring Batch能做什麼

自動化、復雜的大量信息處理，無需用戶交互即可最有效地處理。這些操作通常包括基於時間的事件（例如月末計算、通知或通信）。
定期應用在非常大的數據集上重復處理的復雜業務規則（例如，保險福利確定或費率調整）。
將從內部和外部系統接收的信息集成到記錄系統中，這些信息通常需要以事務方式進行格式化、驗證和處理。批處理用於每天為企業處理數十億筆交易。

業務場景：

定期提交批處理
並發批處理：作業的並行處理
分階段的、企業消息驅動的處理
大規模並行批處理
失敗後手動或計劃重啟
依賴步驟的順序處理（擴展到工作流驅動的批處理）
部分處理：跳過記錄（例如，在回滾時）
整批事務，適用於小批量或現有存儲過程/腳本的情況

總之Spring batch可以做的：

從數據庫、文件或隊列中讀取大量記錄。
以某種方式處理數據。
以修改後的形式寫回數據。

1.3 基礎架構

在這裡插入圖片描述

1.4 核心概念和抽象

在這裡插入圖片描述

核心概念：一個 Job 有一對多的Step，每個步驟都正好有一個 ItemReader、一個ItemProcessor和一個ItemWriter。需要啟動作業（使用 JobLauncher），並且需要存儲有關當前運行進程的元數據（在中 JobRepository）。

2 各個組件介紹

2.1 Job

Job是封裝瞭整個批處理過程的實體。與其他 Spring 項目一樣，一個Job與 XML 配置文件或基於 Java 的配置連接在一起。這種配置可以被稱為“作業配置”。

在這裡插入圖片描述

可配置項：

作業的簡單名稱。
Step實例的定義和排序。
作業是否可重新啟動。

2.2 Step

一個Step是一個域對象，它封裝瞭批處理作業的一個獨立的、連續的階段。因此，每個 Job 完全由一個或多個步驟組成。一個Step包含定義和控制實際批處理所需的所有信息。

在這裡插入圖片描述

一個StepExecution代表一次嘗試執行一個Step。StepExecution 每次Step運行時都會創建一個新的，類似於JobExecution。

2.3 ExecutionContext

一個ExecutionContext表示由框架持久化和控制的鍵/值對的集合，以允許開發人員有一個地方來存儲范圍為StepExecution對象或JobExecution對象的持久狀態。

2.4 JobRepository

JobRepository是上述所有 Stereotypes 的持久性機制。它提供瞭CRUD操作JobLauncher，Job以及Step實現。當 Job第一次啟動，一個JobExecution被從庫中獲得，並且，執行的過程中，StepExecution和JobExecution實施方式是通過將它們傳遞到存儲庫持續。

使用 Java 配置時，@EnableBatchProcessing註解提供瞭一個 JobRepository作為開箱即用自動配置的組件之一。

2.5 JobLauncher

JobLauncher表示一個簡單的接口，用於Job使用給定的集合啟動JobParameters，如以下示例所示：

public interface JobLauncher {

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

期望實現JobExecution從中獲得有效JobRepository並執行Job。

2.6 Item Reader

ItemReader是一種抽象，表示一次檢索Step一個項目的輸入。當ItemReader用完它可以提供的項目時，它通過返回來表明這一點null。

2.7 Item Writer

ItemWriter是一種抽象，表示一次一個Step、一批或一大塊項目的輸出。通常， anItemWriter不知道它接下來應該接收的輸入，並且隻知道在其當前調用中傳遞的項目。

2.8 Item Processor

ItemProcessor是表示項目的業務處理的抽象。當ItemReader讀取一個項目並ItemWriter寫入它們時，它 ItemProcessor提供瞭一個訪問點來轉換或應用其他業務處理。如果在處理該項目時確定該項目無效，則返回 null表示不應寫出該項目。

3 Spring Batch實戰

下面就利用我們所學的理論實現一個最簡單的Spring Batch批處理項目

3.1 依賴和項目結構以及配置文件

依賴

<!--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

3.2 代碼和數據表

數據表

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

Student實體類

/**
 * @desc: Student實體類
 * @author: YanMingXin
 * @create: 2021/10/15-12:17
 **/
@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層

/**
 * @desc: Mapper層
 * @author: YanMingXin
 * @create: 2021/10/15-12:17
 **/
@Mapper
@Repository
public interface StudentDao extends BaseMapper<Student> {
}

模擬數據庫（文件）中讀取類

/**
 * @desc: 模擬數據庫中讀取
 * @author: YanMingXin
 * @create: 2021/10/16-10:13
 **/
public class StudentVirtualDao {

    /**
     * 模擬從數據庫中讀取
     *
     * @return
     */
    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;
    }
}

Service層接口

/**
 * @desc:
 * @author: YanMingXin
 * @create: 2021/10/15-12:16
 **/
public interface StudentService {

    List<Student> selectStudentsFromDB();

    void insertStudent(Student student);
}

Service層實現類

/**
 * @desc: Service層實現類
 * @author: YanMingXin
 * @create: 2021/10/15-12:16
 **/
@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);
    }
}

最核心的配置類BatchConfiguration

/**
 * @desc: BatchConfiguration
 * @author: YanMingXin
 * @create: 2021/10/15-12:25
 **/
@Configuration
@EnableBatchProcessing
@SuppressWarnings("all")
public class BatchConfiguration {

    /**
     * 註入JobBuilderFactory
     */
    @Autowired
    public JobBuilderFactory jobBuilderFactory;

    /**
     * 註入StepBuilderFactory
     */
    @Autowired
    public StepBuilderFactory stepBuilderFactory;

    /**
     * 註入JobRepository
     */
    @Autowired
    public JobRepository jobRepository;

    /**
     * 註入JobLauncher
     */
    @Autowired
    private JobLauncher jobLauncher;

    /**
     * 註入自定義StudentService
     */
    @Autowired
    private StudentService studentService;

    /**
     * 註入自定義job
     */
    @Autowired
    private Job studentJob;

    /**
     * 封裝writer bean
     *
     * @return
     */
    @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;
    }

    /**
     * 封裝reader bean
     *
     * @return
     */
    @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;
    }

    /**
     * 封裝processor bean
     *
     * @return
     */
    @Bean
    public ItemProcessor processor() {
        ItemProcessor processor = new ItemProcessor() {
            @Override
            public Object process(Object o) throws Exception {
                //debug發現o就是reader單次單線程讀取的數據
                return o;
            }
        };
        return processor;
    }

    /**
     * 封裝自定義step
     *
     * @return
     */
    @Bean
    public Step studentStepOne() {
        return stepBuilderFactory.get("studentStepOne")
            .chunk(1)
            .reader(reader()) //加入reader
            .processor(processor())  //加入processor
            .writer(writer())//加入writer
            .build();
    }

    /**
     * 封裝自定義job
     *
     * @return
     */
    @Bean
    public Job studentJob() {
        return jobBuilderFactory.get("studentJob")
            .flow(studentStepOne())//加入step
            .end()
            .build();
    }


    /**
     * 使用spring 定時任務執行
     */
    @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();
        }
    }
}

3.3 測試

在這裡插入圖片描述

項目啟動1s之後

在這裡插入圖片描述

看數據庫，除瞭我們實體類定義的表以外多出來這麼多表，這些表都是spring batch自帶的記錄日志和錯誤的表，具體的字段含義的有待研究

在這裡插入圖片描述

4 實戰後的總結

Spring Batch有非常快的寫入和讀取速度，但是帶來的影響就是非常耗費內存和數據庫連接池的資源如果使用不好的話還會發生異常，因此我們要進行正確的配置，接下來我們進行簡單的源碼探究：

4.1 JobBuilderFactory

job的獲取使用瞭簡單工廠模式和建造者模式JobBuilderFactory獲取JobBuilder在經過配置返回一個job對象的實例，該實例就是Spring Batch中最頂級的組件，包含瞭n和step

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;
   }
}

jobBuilder類

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);
   }
}

4.2 StepBuilderFactory

直接看StepBuilder類

public class StepBuilder extends StepBuilderHelper<StepBuilder> {

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

   /**
    * 用自定義微線程構建步驟，不一定是項處理。  
    */
   public TaskletStepBuilder tasklet(Tasklet tasklet) {
      return new TaskletStepBuilder(this).tasklet(tasklet);
   }

   /**
    * 構建一個步驟，按照提供的大小以塊的形式處理項。為瞭將這一步擴展到容錯，
    * 在構建器上調用SimpleStepBuilder的 faultolerant()方法。
    * @param <I> 輸入類型
    * @param <O> 輸出類型
    */
   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);
   }
}

參考文檔：

https://docs.spring.io/spring-batch/docs/4.3.x/reference/html/index.html

https://www.jdon.com/springbatch.html

到此這篇關於Spring Batch輕量級批處理框架實戰的文章就介紹到這瞭,更多相關Spring Batch批處理內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

Spring Batch輕量級批處理框架實戰

目錄

1 實戰前的理論基礎

1.1 Spring Batch是什麼

1.2 Spring Batch能做什麼

1.3 基礎架構

1.4 核心概念和抽象

2 各個組件介紹

2.1 Job

2.2 Step

2.3 ExecutionContext

2.4 JobRepository

2.5 JobLauncher

2.6 Item Reader

2.7 Item Writer

2.8 Item Processor

3 Spring Batch實戰

3.1 依賴和項目結構以及配置文件

3.2 代碼和數據表

3.3 測試

4 實戰後的總結

4.1 JobBuilderFactory

4.2 StepBuilderFactory

參考文檔：

推薦閱讀：

發佈留言取消回覆

近期文章

目錄

1 實戰前的理論基礎

1.1 Spring Batch是什麼

1.2 Spring Batch能做什麼

1.3 基礎架構

1.4 核心概念和抽象

2 各個組件介紹

2.1 Job

2.2 Step

2.3 ExecutionContext

2.4 JobRepository

2.5 JobLauncher

2.6 Item Reader

2.7 Item Writer

2.8 Item Processor

3 Spring Batch實戰

3.1 依賴和項目結構以及配置文件

3.2 代碼和數據表

3.3 測試

4 實戰後的總結

4.1 JobBuilderFactory

4.2 StepBuilderFactory

參考文檔：

推薦閱讀：

發佈留言 取消回覆

近期文章

標籤

發佈留言取消回覆