摘要：什么是作為的子項(xiàng)目，是一款基于的企業(yè)批處理框架。首先，運(yùn)行的基本單位是一個(gè)，一個(gè)就做一件批處理的事情?？偨Y(jié)為我們提供了非常實(shí)用的功能，對(duì)批處理場(chǎng)景進(jìn)行了完善的抽象，它不僅能實(shí)現(xiàn)小數(shù)據(jù)的遷移，也能應(yīng)對(duì)大企業(yè)的大數(shù)據(jù)實(shí)踐應(yīng)用。

本文將從0到1講解一個(gè)Spring Batch是如何搭建并運(yùn)行起來(lái)的。
本教程將講解從一個(gè)文本文件讀取數(shù)據(jù)，然后寫入MySQL。

Spring Batch 作為 Spring 的子項(xiàng)目，是一款基于 Spring 的企業(yè)批處理框架。通過(guò)它可以構(gòu)建出健壯的企業(yè)批處理應(yīng)用。Spring Batch 不僅提供了統(tǒng)一的讀寫接口、豐富的任務(wù)處理方式、靈活的事務(wù)管理及并發(fā)處理，同時(shí)還支持日志、監(jiān)控、任務(wù)重啟與跳過(guò)等特性，大大簡(jiǎn)化了批處理應(yīng)用開發(fā)，將開發(fā)人員從復(fù)雜的任務(wù)配置管理過(guò)程中解放出來(lái)，使他們可以更多地去關(guān)注核心的業(yè)務(wù)處理過(guò)程。

更多的介紹可以參考官網(wǎng)：https://spring.io/projects/sp...

我是用的Intellij Idea，用gradle構(gòu)建。

可以使用Spring Initializr 來(lái)創(chuàng)建Spring boot應(yīng)用。地址：https://start.spring.io/

首先選擇Gradle Project，然后選擇Java。填上你的Group和Artifact名字。

最后再搜索你需要用的包，比如Batch是一定要的。另外，由于我寫的Batch項(xiàng)目是使用JPA向MySQL插入數(shù)據(jù)，所以也添加了JPA和MySQL。其他可以根據(jù)自己需要添加。

點(diǎn)擊Generate Project，一個(gè)項(xiàng)目就創(chuàng)建好了。

Build.gralde文件大概就長(zhǎng)這個(gè)樣子：

buildscript {
   ext {
      springBootVersion = "2.0.4.RELEASE"
   }
   repositories {
      mavenCentral()
   }
   dependencies {
      classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}")
   }
}

apply plugin: "java"
apply plugin: "idea"
apply plugin: "org.springframework.boot"
apply plugin: "io.spring.dependency-management"

group = "com.demo"
version = "0.0.1-SNAPSHOT"
sourceCompatibility = 1.8

repositories {
   mavenCentral()
}

dependencies {
   compile("org.springframework.boot:spring-boot-starter-batch")
   compile("org.springframework.boot:spring-boot-starter-jdbc")
   compile("org.springframework.boot:spring-boot-starter-data-jpa")
   compile group: "com.fasterxml.jackson.datatype", name: "jackson-datatype-joda", version: "2.9.4"
   compile group: "org.jadira.usertype", name: "usertype.core", version: "6.0.1.GA"
   compile group: "mysql", name: "mysql-connector-java", version: "6.0.6",
   testCompile("org.springframework.boot:spring-boot-starter-test")
   testCompile("org.springframework.batch:spring-batch-test")
}

網(wǎng)上有很多Spring Batch結(jié)構(gòu)和原理的講解，我就不詳細(xì)闡述了，我這里只講一下Spring Batch的一個(gè)基本層級(jí)結(jié)構(gòu)。

首先，Spring Batch運(yùn)行的基本單位是一個(gè)Job，一個(gè)Job就做一件批處理的事情。
一個(gè)Job包含很多Step，step就是每個(gè)job要執(zhí)行的單個(gè)步驟。

如下圖所示，Step里面，會(huì)有Tasklet，Tasklet是一個(gè)任務(wù)單元，它是屬于可以重復(fù)利用的東西。
然后是Chunk，chunk就是數(shù)據(jù)塊，你需要定義多大的數(shù)據(jù)量是一個(gè)chunk。

Chunk里面就是不斷循環(huán)的一個(gè)流程，讀數(shù)據(jù)，處理數(shù)據(jù)，然后寫數(shù)據(jù)。Spring Batch會(huì)不斷的循環(huán)這個(gè)流程，直到批處理數(shù)據(jù)完成。

首先，我們需要一個(gè)全局的Configuration來(lái)配置所有的Job和一些全局配置。

代碼如下：

@Configuration
@EnableAutoConfiguration
@EnableBatchProcessing(modular = true)
public class SpringBatchConfiguration {
    @Bean
    public ApplicationContextFactory firstJobContext() {
        return new GenericApplicationContextFactory(FirstJobConfiguration.class);
    }
    
    @Bean
    public ApplicationContextFactory secondJobContext() {
        return new GenericApplicationContextFactory(SecondJobConfiguration.class);
    }

}

@EnableBatchProcessing是打開Batch。如果要實(shí)現(xiàn)多Job的情況，需要把EnableBatchProcessing注解的modular設(shè)置為true，讓每個(gè)Job使用自己的ApplicationConext。

比如上面代碼的就創(chuàng)建了兩個(gè)Job。

本博客的例子是遷移數(shù)據(jù)，數(shù)據(jù)源是一個(gè)文本文件，數(shù)據(jù)量是上百萬(wàn)條，一行就是一條數(shù)據(jù)。然后我們通過(guò)Spring Batch幫我們把文本文件的數(shù)據(jù)全部遷移到MySQL數(shù)據(jù)庫(kù)對(duì)應(yīng)的表里面。

假設(shè)我們遷移的數(shù)據(jù)是Message，那么我們就需要提前創(chuàng)建一個(gè)叫Message的和數(shù)據(jù)庫(kù)映射的數(shù)據(jù)類。

@Entity
@Table(name = "message")
public class Message {
    @Id
    @Column(name = "object_id", nullable = false)
    private String objectId;

    @Column(name = "content")
    private String content;

    @Column(name = "last_modified_time")
    private LocalDateTime lastModifiedTime;

    @Column(name = "created_time")
    private LocalDateTime createdTime;
}

首先我們需要一個(gè)關(guān)于這個(gè)Job的Configuration，它將在SpringBatchConfigration里面被加載。

@Configuration
@EnableAutoConfiguration
@EnableBatchProcessing(modular = true)
public class SpringBatchConfiguration {
    @Bean
    public ApplicationContextFactory messageMigrationJobContext() {
        return new GenericApplicationContextFactory(MessageMigrationJobConfiguration.class);
    }
}

下面的關(guān)于構(gòu)建Job的代碼都將寫在這個(gè)MessageMigrationJobConfiguration里面。

public class MessageMigrationJobConfiguration {
}

我們先定義一個(gè)Job的Bean。

@Autowired
private JobBuilderFactory jobBuilderFactory;

@Bean
public Job messageMigrationJob(@Qualifier("messageMigrationStep") Step messageMigrationStep) {
    return jobBuilderFactory.get("messageMigrationJob")
            .start(messageMigrationStep)
            .build();
}

jobBuilderFactory是注入進(jìn)來(lái)的，get里面的就是job的名字。
這個(gè)job只有一個(gè)step。

接下來(lái)就是創(chuàng)建Step。

@Autowired
private StepBuilderFactory stepBuilderFactory;

@Bean
public Step messageMigrationStep(@Qualifier("jsonMessageReader") FlatFileItemReader jsonMessageReader,
                                 @Qualifier("messageItemWriter") JpaItemWriter messageItemWriter,
                                 @Qualifier("errorWriter") Writer errorWriter) {
    return stepBuilderFactory.get("messageMigrationStep")
            .chunk(CHUNK_SIZE)
            .reader(jsonMessageReader).faultTolerant().skip(JsonParseException.class).skipLimit(SKIP_LIMIT)
            .listener(new MessageItemReadListener(errorWriter))
            .writer(messageItemWriter).faultTolerant().skip(Exception.class).skipLimit(SKIP_LIMIT)
            .listener(new MessageWriteListener())
            .build();
}

stepBuilderFactory是注入進(jìn)來(lái)的，然后get里面是Step的名字。
我們的Step中可以構(gòu)建很多東西，比如reader，processer，writer，listener等等。

下面我們就逐個(gè)來(lái)看看step里面的這些東西是如何使用的。

Spring batch在配置Step時(shí)采用的是基于Chunk的機(jī)制，即每次讀取一條數(shù)據(jù)，再處理一條數(shù)據(jù)，累積到一定數(shù)量后再一次性交給writer進(jìn)行寫入操作。這樣可以最大化的優(yōu)化寫入效率，整個(gè)事務(wù)也是基于Chunk來(lái)進(jìn)行。

比如我們定義chunk size是50，那就意味著，spring batch處理了50條數(shù)據(jù)后，再統(tǒng)一向數(shù)據(jù)庫(kù)寫入。
這里有個(gè)很重要的點(diǎn)，chunk前面需要定義數(shù)據(jù)輸入類型和輸出類型，由于我們輸入是Message，輸出也是Message，所以兩個(gè)都直接寫Message了。
如果不定義這個(gè)類型，會(huì)報(bào)錯(cuò)。

.chunk(CHUNK_SIZE)

Reader顧名思義就是從數(shù)據(jù)源讀取數(shù)據(jù)。
Spring Batch給我們提供了很多好用實(shí)用的reader，基本能滿足我們所有需求。比如FlatFileItemReader，JdbcCursorItemReader，JpaPagingItemReader等。也可以自己實(shí)現(xiàn)Reader。

本例子里面，數(shù)據(jù)源是文本文件，所以我們就使用FlatFileItemReader。FlatFileItemReader是從文件里面一行一行的讀取數(shù)據(jù)。
首先需要設(shè)置文件路徑，也就是設(shè)置resource。
因?yàn)槲覀冃枰岩恍形谋居成錇镸essage類，所以我們需要自己設(shè)置并實(shí)現(xiàn)LineMapper。

@Bean
public FlatFileItemReader jsonMessageReader() {
    FlatFileItemReader reader = new FlatFileItemReader<>();
    reader.setResource(new FileSystemResource(new File(MESSAGE_FILE)));
    reader.setLineMapper(new MessageLineMapper());
    return reader;
}

LineMapper的輸入就是獲取一行文本，和行號(hào)，然后轉(zhuǎn)換成Message。
在本例子里面，一行文本就是一個(gè)json對(duì)象，所以我們使用JsonParser來(lái)轉(zhuǎn)換成Message。

public class MessageLineMapper implements LineMapper {
    private MappingJsonFactory factory = new MappingJsonFactory();

    @Override
    public Message mapLine(String line, int lineNumber) throws Exception {   
        JsonParser parser = factory.createParser(line);
        Map map = (Map) parser.readValueAs(Map.class);
        Message message = new Message();
        ... // 轉(zhuǎn)換邏輯
        return message;
    }
}

由于本例子里面，數(shù)據(jù)是一行文本，通過(guò)reader變成Message的類，然后writer直接把Message寫入MySQL。所以我們的例子里面就不需要Processor，關(guān)于如何寫Processor其實(shí)和reader/writer是一樣的道理。
從它的接口可以看出，需要定義輸入和輸出的類型，把輸入I通過(guò)某些邏輯處理之后，返回輸出O。

public interface ItemProcessor {
    O process(I item) throws Exception;
}

Writer顧名思義就是把數(shù)據(jù)寫入到目標(biāo)數(shù)據(jù)源里面。
Spring Batch同樣給我們提供很多好用實(shí)用的writer。比如JpaItemWriter，F(xiàn)latFileItemWriter，HibernateItemWriter，JdbcBatchItemWriter等。同樣也可以自定義。

本例子里面，使用的是JpaItemWriter，可以直接把Message對(duì)象寫到數(shù)據(jù)庫(kù)里面。但是需要設(shè)置一個(gè)EntityManagerFactory，可以注入進(jìn)來(lái)。

@Autowired
private EntityManagerFactory entityManager;

@Bean
public JpaItemWriter messageItemWriter() {
    JpaItemWriter writer = new JpaItemWriter<>();
    writer.setEntityManagerFactory(entityManager);
    return writer;
}

另外，你需要配置數(shù)據(jù)庫(kù)的連接等東西。由于我使用的spring，所以直接在Application.properties里面配置如下：

spring.datasource.url=jdbc:mysql://database
spring.datasource.username=username
spring.datasource.password=password
spring.datasource.driverClassName=com.mysql.cj.jdbc.Driver
spring.jpa.database-platform=org.hibernate.dialect.MySQLDialect
spring.jpa.show-sql=true
spring.jpa.properties.jadira.usertype.autoRegisterUserTypes=true
spring.jackson.serialization.write-dates-as-timestamps=false
spring.batch.initialize-schema=ALWAYS
spring.jpa.hibernate.ddl-auto=update

spring.datasource相關(guān)的設(shè)置都是在配置數(shù)據(jù)庫(kù)的連接。
spring.batch.initialize-schema=always表示讓spring batch在數(shù)據(jù)庫(kù)里面創(chuàng)建默認(rèn)的數(shù)據(jù)表。
spring.jpa.show-sql=true表示在控制臺(tái)輸出hibernate讀寫數(shù)據(jù)庫(kù)時(shí)候的SQL。
spring.jpa.database-platform=org.hibernate.dialect.MySQLDialect是在指定MySQL的方言。

Spring Batch同樣實(shí)現(xiàn)了非常完善全面的listener，listener很好理解，就是用來(lái)監(jiān)聽每個(gè)步驟的結(jié)果。比如可以有監(jiān)聽step的，有監(jiān)聽job的，有監(jiān)聽reader的，有監(jiān)聽writer的。沒(méi)有你找不到的listener，只有你想不到的listener。

在本例子里面，我只關(guān)心，read的時(shí)候有沒(méi)有出錯(cuò)，和write的時(shí)候有沒(méi)有出錯(cuò)，所以，我只實(shí)現(xiàn)了ReadListener和WriteListener。
在read出錯(cuò)的時(shí)候，把錯(cuò)誤結(jié)果寫入一個(gè)多帶帶的error列表文件中。

public class MessageItemReadListener implements ItemReadListener {
    private Writer errorWriter;

    public MessageItemReadListener(Writer errorWriter) {
        this.errorWriter = errorWriter;
    }

    @Override
    public void beforeRead() {
    }

    @Override
    public void afterRead(Message item) {
    }

    @Override
    public void onReadError(Exception ex) {
         errorWriter.write(format("%s%n", ex.getMessage()));
    }
}

在write出錯(cuò)的時(shí)候，也做同樣的事情，把出錯(cuò)的原因?qū)懭攵鄮У娜罩局小?/p>

public class MessageWriteListener implements ItemWriteListener {

    @Autowired
    private Writer errorWriter;

    @Override
    public void beforeWrite(List items) {
    }

    @Override
    public void afterWrite(List items) {
    }

    @Override
    public void onWriteError(Exception exception, List items) {
        errorWriter.write(format("%s%n", exception.getMessage()));
        for (Message message : items) {
            errorWriter.write(format("Failed writing message id: %s", message.getObjectId()));
        }
    }
}

前面有說(shuō)chuck機(jī)制，所以write的listener傳入?yún)?shù)是一個(gè)List，因?yàn)樗抢鄯e到一定的數(shù)量才一起寫入。

Spring Batch提供了skip的機(jī)制，也就是說(shuō)，如果出錯(cuò)了，可以跳過(guò)。如果你不設(shè)置skip，那么一條數(shù)據(jù)出錯(cuò)了，整個(gè)job都會(huì)掛掉。
設(shè)置skip的時(shí)候一定要設(shè)置什么Exception才需要跳過(guò)，并且跳過(guò)多少條數(shù)據(jù)。如果失敗的數(shù)據(jù)超過(guò)你設(shè)置的skip limit，那么job就會(huì)失敗。
你可以分別給reader和writer等設(shè)置skip機(jī)制。

writer(messageItemWriter).faultTolerant().skip(Exception.class).skipLimit(SKIP_LIMIT)

這個(gè)和Skip是一樣的原理，就是失敗之后可以重試，你同樣需要設(shè)置重試的次數(shù)。
同樣可以分別給reader，writer等設(shè)置retry機(jī)制。

如果同時(shí)設(shè)置了retry和skip，會(huì)先重試所有次數(shù)，然后再開始skip。比如retry是10次，skip是20，會(huì)先重試10次之后，再開始算第一次skip。

所有東西都準(zhǔn)備好以后，就是如何運(yùn)行了。
運(yùn)行就是在main方法里面用JobLauncher去運(yùn)行你制定的job。

下面是我寫的main方法，main方法的第一個(gè)參數(shù)是job的名字，這樣我們就可以通過(guò)不同的job名字跑不同的job了。

首先我們通過(guò)運(yùn)行起來(lái)的Spring application得到j(luò)obRegistry，然后通過(guò)job的名字找到對(duì)應(yīng)的job。

接著，我們就可以用jobLauncher去運(yùn)行這個(gè)job了，運(yùn)行的時(shí)候會(huì)傳一些參數(shù)，比如你job里面需要的文件路徑或者文件日期等，就可以通過(guò)這個(gè)jobParameters傳進(jìn)去。如果沒(méi)有參數(shù)，可以默認(rèn)傳當(dāng)前時(shí)間進(jìn)去。

public static void main(String[] args) {
    String jobName = args[0];

    try {
        ConfigurableApplicationContext context = SpringApplication.run(ZuociBatchApplication.class, args);
        JobRegistry jobRegistry = context.getBean(JobRegistry.class);
        Job job = jobRegistry.getJob(jobName);
        JobLauncher jobLauncher = context.getBean(JobLauncher.class);
        JobExecution jobExecution = jobLauncher.run(job, createJobParams());
        if (!jobExecution.getExitStatus().equals(ExitStatus.COMPLETED)) {
            throw new RuntimeException(format("%s Job execution failed.", jobName));
        }
    } catch (Exception e) {
        throw new RuntimeException(format("%s Job execution failed.", jobName));
    }
}

private static JobParameters createJobParams() {
    return new JobParametersBuilder().addDate("date", new Date()).toJobParameters();
}

最后，把jar包編譯出來(lái)，在命令行執(zhí)行下面的命令，就可以運(yùn)行你的Spring Batch了。

java -jar YOUR_BATCH_NAME.jar YOUR_JOB_NAME

調(diào)試主要依靠控制臺(tái)輸出的log，可以在application.properties里面設(shè)置log輸出的級(jí)別，比如你希望輸出INFO信息還是DEBUG信息。
基本上，通過(guò)查看log都能定位到問(wèn)題。

logging.path=build/logs
logging.file=${logging.path}/batch.log
logging.level.com.easystudio=INFO
logging.level.root=INFO
log4j.logger.org.springframework.jdbc=INFO
log4j.logger.org.springframework.batch=INFO
logging.level.org.hibernate.SQL=INFO

如果你的batch最終會(huì)寫入數(shù)據(jù)庫(kù)，那么Spring Batch會(huì)默認(rèn)在你的數(shù)據(jù)庫(kù)里面創(chuàng)建一些batch相關(guān)的表，來(lái)記錄所有job/step運(yùn)行的狀態(tài)和結(jié)果。

大部分表你都不需要關(guān)心，你只需要關(guān)心幾張表。

batch_job_instance：這張表能看到每次運(yùn)行的job名字。

batch_job_execution：這張表能看到每次運(yùn)行job的開始時(shí)間，結(jié)束時(shí)間，狀態(tài)，以及失敗后的錯(cuò)誤消息是什么。

batch_step_execution：這張表你能看到更多關(guān)于step的詳細(xì)信息。比如step的開始時(shí)間，結(jié)束時(shí)間，提交次數(shù)，讀寫次數(shù)，狀態(tài)，以及失敗后的錯(cuò)誤信息等。

Spring Batch為我們提供了非常實(shí)用的功能，對(duì)批處理場(chǎng)景進(jìn)行了完善的抽象，它不僅能實(shí)現(xiàn)小數(shù)據(jù)的遷移，也能應(yīng)對(duì)大企業(yè)的大數(shù)據(jù)實(shí)踐應(yīng)用。它讓我們開發(fā)批處理應(yīng)用可以事半功倍。

最后一個(gè)tips，搭建Spring Batch的過(guò)程中，會(huì)遇到各種各樣的問(wèn)題。只要善用Google，都能找到答案。