創(chuàng)建批處理服務(wù)

本指南將引導(dǎo)您完成創(chuàng)建基本批處理驅(qū)動解決方案的過程。

你將建造什么

您將構(gòu)建一個從 CSV 電子表格導(dǎo)入數(shù)據(jù)、使用自定義代碼對其進(jìn)行轉(zhuǎn)換并將最終結(jié)果存儲在數(shù)據(jù)庫中的服務(wù)。

你需要什么

1. 約15分鐘
2. 最喜歡的文本編輯器或 IDE
3. JDK 1.8或更高版本
4. Gradle 4+或Maven 3.2+
5. 您還可以將代碼直接導(dǎo)入 IDE：

• 彈簧工具套件 (STS)
• IntelliJ IDEA

如何完成本指南

像大多數(shù) Spring入門指南一樣，您可以從頭開始并完成每個步驟，也可以繞過您已經(jīng)熟悉的基本設(shè)置步驟。無論哪種方式，您最終都會得到工作代碼。

要從頭開始，請繼續(xù)從 Spring Initializr 開始。

要跳過基礎(chǔ)知識，請執(zhí)行以下操作：

• 下載并解壓本指南的源代碼庫，或使用Git克隆它：git clone https://github.com/spring-guides/gs-batch-processing.git
• 光盤進(jìn)入gs-batch-processing/initial
• 繼續(xù)創(chuàng)建商務(wù)艙。

完成后，您可以對照中的代碼檢查結(jié)果

gs-batch-processing/complete。

業(yè)務(wù)數(shù)據(jù)

通常，您的客戶或業(yè)務(wù)分析師會提供電子表格。對于這個簡單的示例，您可以在以下位置找到一些虛構(gòu)的數(shù)據(jù)src/main/resources/sample-data.csv：

Jill,Doe
Joe,Doe
Justin,Doe
Jane,Doe
John,Doe

此電子表格的每一行都包含名字和姓氏，以逗號分隔。這是一種相當(dāng)常見的模式，Spring 無需定制即可處理。

接下來，您需要編寫一個 SQL 腳本來創(chuàng)建一個表來存儲數(shù)據(jù)。您可以在以下位置找到這樣的腳本src/main/resources/schema-all.sql：

DROP TABLE people IF EXISTS; CREATE TABLE people ( person_id BIGINT IDENTITY NOT NULL PRIMARY KEY, first_name VARCHAR(20), last_name VARCHAR(20) ); 

Spring Bootschema-@@platform@@.sql在啟動期間自動運行。-all是所有平臺的默認(rèn)設(shè)置。

從 Spring Initializr 開始

您可以使用這個預(yù)先初始化的項目并單擊 Generate 下載 ZIP 文件。此項目配置為適合本教程中的示例。

手動初始化項目：

1. 導(dǎo)航到https://start.spring.io。該服務(wù)提取應(yīng)用程序所需的所有依賴項，并為您完成大部分設(shè)置。
2. 選擇 Gradle 或 Maven 以及您要使用的語言。本指南假定您選擇了 Java。
3. 單擊Dependencies并選擇Spring Batch和HyperSQL Database。
4. 單擊生成。
5. 下載生成的 ZIP 文件，該文件是根據(jù)您的選擇配置的 Web 應(yīng)用程序的存檔。

如果您的 IDE 具有 Spring Initializr 集成，您可以從您的 IDE 完成此過程。

你也可以從 Github 上 fork 項目并在你的 IDE 或其他編輯器中打開它。

創(chuàng)建商務(wù)艙

現(xiàn)在您可以看到數(shù)據(jù)輸入和輸出的格式，您可以編寫代碼來表示一行數(shù)據(jù)，如以下示例(來自src/main/java/com/example/batchprocessing/Person.java)所示：

package com.example.batchprocessing; public class Person { private String lastName; private String firstName; public Person() { } public Person(String firstName, String lastName) { this.firstName = firstName; this.lastName = lastName; } public void setFirstName(String firstName) { this.firstName = firstName; } public String getFirstName() { return firstName; } public String getLastName() { return lastName; } public void setLastName(String lastName) { this.lastName = lastName; } @Override
public String toString() { return "firstName: " + firstName + ", lastName: " + lastName; } } 

創(chuàng)建中間處理器

批處理中的一個常見范例是攝取數(shù)據(jù)，對其進(jìn)行轉(zhuǎn)換，然后將其通過管道輸出到其他地方。在這里，您需要編寫一個簡單的轉(zhuǎn)換器，將名稱轉(zhuǎn)換為大寫。以下清單(來自src/main/java/com/example/batchprocessing/PersonItemProcessor.java)顯示了如何執(zhí)行此操作：

package com.example.batchprocessing; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.batch.item.ItemProcessor; public class PersonItemProcessor implements ItemProcessor { private static final Logger log = LoggerFactory.getLogger(PersonItemProcessor.class); @Override
public Person process(final Person person) throws Exception { final String firstName = person.getFirstName().toUpperCase(); final String lastName = person.getLastName().toUpperCase(); final Person transformedPerson = new Person(firstName, lastName); log.info("Converting (" + person + ") into (" + transformedPerson + ")"); return transformedPerson; } } 

PersonItemProcessor實現(xiàn) Spring Batch 的ItemProcessor接口。這使得將代碼連接到您將在本指南后面定義的批處理作業(yè)變得很容易。根據(jù)界面，您會收到一個傳入的Person對象，然后將其轉(zhuǎn)換為大寫的Person.

輸入和輸出類型不必相同。事實上，在讀取一個數(shù)據(jù)源之后，有時應(yīng)用程序的數(shù)據(jù)流需要不同的數(shù)據(jù)類型。

將批處理作業(yè)放在一起

現(xiàn)在您需要將實際的批處理作業(yè)放在一起。Spring Batch 提供了許多實用程序類來減少編寫自定義代碼的需要。相反，您可以專注于業(yè)務(wù)邏輯。

要配置您的作業(yè)，您必須首先創(chuàng)建一個 Spring@Configuration類，如下例所示src/main/java/com/exampe/batchprocessing/BatchConfiguration.java：

@Configuration
@EnableBatchProcessing
public class BatchConfiguration { @Autowired
public JobBuilderFactory jobBuilderFactory; @Autowired
public StepBuilderFactory stepBuilderFactory; ... } 

對于初學(xué)者，@EnableBatchProcessing注釋添加了許多支持作業(yè)并為您節(jié)省大量工作的關(guān)鍵 bean。此示例使用基于內(nèi)存的數(shù)據(jù)庫(由 提供@EnableBatchProcessing)，這意味著完成后，數(shù)據(jù)就消失了。它還自動連接下面需要的幾個工廠。現(xiàn)在將以下 bean 添加到您的BatchConfiguration類中以定義讀取器、處理器和寫入器：

@Bean
public FlatFileItemReader reader() { return new FlatFileItemReaderBuilder() .name("personItemReader") .resource(new ClassPathResource("sample-data.csv")) .delimited() .names(new String[]{"firstName", "lastName"}) .fieldSetMapper(new BeanWrapperFieldSetMapper() {{ setTargetType(Person.class); }}) .build(); } @Bean
public PersonItemProcessor processor() { return new PersonItemProcessor(); } @Bean
public JdbcBatchItemWriter writer(DataSource dataSource) { return new JdbcBatchItemWriterBuilder() .itemSqlParameterSourceProvider(new 
BeanPropertyItemSqlParameterSourceProvider<>()) .sql("INSERT INTO people (first_name, last_name) VALUES (:firstName,  :lastName)") .dataSource(dataSource) .build(); } 

第一段代碼定義了輸入、處理器和輸出。

• reader()創(chuàng)建一個ItemReader. 它查找一個名為的文件sample-data.csv，并使用足夠的信息解析每個行項目，以將其轉(zhuǎn)換為Person.
• processor()創(chuàng)建一個PersonItemProcessor您之前定義的實例，用于將數(shù)據(jù)轉(zhuǎn)換為大寫。
• writer(DataSource)創(chuàng)建一個ItemWriter. 這個針對 JDBC 目標(biāo)，并自動獲取由@EnableBatchProcessing. 它包括插入單個 所需的 SQL 語句Person，由 Java bean 屬性驅(qū)動。

最后一個塊(來自src/main/java/com/example/batchprocessing/BatchConfiguration.java)顯示了實際的作業(yè)配置：

@Bean
public Job importUserJob(JobCompletionNotificationListener listener, Step 
step1) { return jobBuilderFactory.get("importUserJob") .incrementer(new RunIdIncrementer()) .listener(listener) .flow(step1) .end() .build(); } @Bean
public Step step1(JdbcBatchItemWriter writer) { return stepBuilderFactory.get("step1") .reader(reader()) .processor(processor()) .writer(writer) .build(); } 

第一種方法定義了作業(yè)，第二種方法定義了一個步驟。作業(yè)是由步驟構(gòu)建的，其中每個步驟都可能涉及讀取器、處理器和寫入器。

在此作業(yè)定義中，您需要一個增量器，因為作業(yè)使用數(shù)據(jù)庫來維護(hù)執(zhí)行狀態(tài)。然后列出每個步驟(盡管此作業(yè)只有一個步驟)。作業(yè)結(jié)束，Java API 生成一個完美配置的作業(yè)。

在步驟定義中，您定義一次寫入多少數(shù)據(jù)。在這種情況下，它一次最多寫入十個記錄。接下來，您使用之前注入的 bean 配置讀取器、處理器和寫入器。

chunk()是前綴，因為它是一個通用方法。這表示每個處理“塊”的輸入和輸出類型，并與ItemReader和對齊ItemWriter。

批處理配置的最后一點是在作業(yè)完成時獲得通知的一種方式。以下示例(來自src/main/java/com/example/batchprocessing/JobCompletionNotificationListener.java)顯示了這樣一個類：

package com.example.batchprocessing; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.batch.core.BatchStatus; import org.springframework.batch.core.JobExecution; import 
org.springframework.batch.core.listener.JobExecutionListenerSupport; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.stereotype.Component; @Component
public class JobCompletionNotificationListener extends 
JobExecutionListenerSupport { private static final Logger log = LoggerFactory.getLogger(JobCompletionNotificationListener.class); private final JdbcTemplate jdbcTemplate; @Autowired
public JobCompletionNotificationListener(JdbcTemplate jdbcTemplate) { this.jdbcTemplate = jdbcTemplate; } @Override
public void afterJob(JobExecution jobExecution) { if(jobExecution.getStatus() == BatchStatus.COMPLETED) { log.info("!!! JOB FINISHED! Time to verify the results"); jdbcTemplate.query("SELECT first_name, last_name FROM people", (rs, row) -> new Person( rs.getString(1), rs.getString(2)) ).forEach(person -> log.info("Found <" + person + "> in the  database.")); } } } 

JobCompletionNotificationListener監(jiān)聽作業(yè)的時間，BatchStatus.COMPLETED然后用于JdbcTemplate檢查結(jié)果。

使應(yīng)用程序可執(zhí)行

盡管批處理可以嵌入到 Web 應(yīng)用程序和 WAR 文件中，但下面演示的更簡單的方法可以創(chuàng)建一個獨立的應(yīng)用程序。您將所有內(nèi)容打包在一個可執(zhí)行的 JAR 文件中，由一個很好的舊 Javamain()方法驅(qū)動。

Spring Initializr 為您創(chuàng)建了一個應(yīng)用程序類。對于這個簡單的示例，它無需進(jìn)一步修改即可工作。以下清單(來自

src/main/java/com/example/batchprocessing/BatchProcessingApplication.java)顯示了應(yīng)用程序類：

JobCompletionNotificationListener監(jiān)聽作業(yè)的時間，BatchStatus.COMPLETED然后用于JdbcTemplate檢查結(jié)果。

使應(yīng)用程序可執(zhí)行

盡管批處理可以嵌入到 Web 應(yīng)用程序和 WAR 文件中，但下面演示的更簡單的方法可以創(chuàng)建一個獨立的應(yīng)用程序。您將所有內(nèi)容打包在一個可執(zhí)行的 JAR 文件中，由一個很好的舊 Javamain()方法驅(qū)動。

Spring Initializr 為您創(chuàng)建了一個應(yīng)用程序類。對于這個簡單的示例，它無需進(jìn)一步修改即可工作。以下清單(來自

src/main/java/com/example/batchprocessing/BatchProcessingApplication.java)顯示了應(yīng)用程序類：

package com.example.batchprocessing; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication
public class BatchProcessingApplication { public static void main(String[] args) throws Exception { System.exit(SpringApplication.exit(SpringApplication.run(BatchProcessingApplication.class, args))); } } 

@SpringBootApplication是一個方便的注釋，它添加了以下所有內(nèi)容：

• @Configuration: 將類標(biāo)記為應(yīng)用程序上下文的 bean 定義源。
• @EnableAutoConfiguration：告訴 Spring Boot 根據(jù)類路徑設(shè)置、其他 bean 和各種屬性設(shè)置開始添加 bean。例如，如果spring-webmvc位于類路徑上，則此注釋將應(yīng)用程序標(biāo)記為 Web 應(yīng)用程序并激活關(guān)鍵行為，例如設(shè)置DispatcherServlet.
• @ComponentScan: 告訴 Spring 在包中查找其他組件、配置和服務(wù)com/example，讓它找到控制器。

該main()方法使用 Spring Boot 的SpringApplication.run()方法來啟動應(yīng)用程序。您是否注意到?jīng)]有一行 XML?也沒有web.xml文件。這個 Web 應(yīng)用程序是 100% 純 Java，您不必處理任何管道或基礎(chǔ)設(shè)施的配置。

請注意SpringApplication.exit()并System.exit()確保 JVM 在作業(yè)完成后退出。有關(guān)更多詳細(xì)信息，請參閱Spring Boot 參考文檔中的應(yīng)用程序退出部分。

出于演示目的，有代碼可以創(chuàng)建一個JdbcTemplate、查詢數(shù)據(jù)庫并打印出批處理作業(yè)插入的人員姓名。

構(gòu)建一個可執(zhí)行的 JAR

您可以使用 Gradle 或 Maven 從命令行運行應(yīng)用程序。您還可以構(gòu)建一個包含所有必要依賴項、類和資源的單個可執(zhí)行 JAR 文件并運行它。構(gòu)建可執(zhí)行 jar 可以在整個開發(fā)生命周期、跨不同環(huán)境等中輕松地作為應(yīng)用程序交付、版本化和部署服務(wù)。

如果您使用 Gradle，則可以使用./gradlew bootRun. 或者，您可以使用構(gòu)建 JAR 文件./gradlew build，然后運行 JAR 文件，如下所示：

java -jar build/libs/gs-batch-processing-0.1.0.jar 

如果您使用 Maven，則可以使用./mvnw spring-boot:run. 或者，您可以使用構(gòu)建 JAR 文件，./mvnw clean package然后運行該 JAR 文件，如下所示：

java -jar 目標(biāo)/gs-batch-processing-0.1.0.jar 

此處描述的步驟創(chuàng)建了一個可運行的 JAR。您還可以構(gòu)建經(jīng)典的 WAR 文件。

該作業(yè)為每個被轉(zhuǎn)換的人打印一行。作業(yè)運行后，您還可以看到查詢數(shù)據(jù)庫的輸出。它應(yīng)該類似于以下輸出：

java -jar 目標(biāo)/gs-batch-processing-0.1.0.jar 

概括

恭喜!您構(gòu)建了一個批處理作業(yè)，該作業(yè)從電子表格中提取數(shù)據(jù)，對其進(jìn)行處理，然后將其寫入數(shù)據(jù)庫。

原文地址：https://www.toutiao.com/a7070000624930259459/