java批處理 異常處理

在當(dāng)今世界，互聯(lián)網(wǎng)已經(jīng)改變了我們的生活方式，其主要原因之一是大部分日?，嵤露际褂没ヂ?lián)網(wǎng)。 這導(dǎo)致大量數(shù)據(jù)可用于處理。

其中涉及大量數(shù)據(jù)的一些示例是處理工資單，銀行對帳單，利息計(jì)算等。因此，想象一下，如果所有這些工作都必須手動完成，那么完成這些工作將花費(fèi)很多時(shí)間。

在目前的年齡如何？ 答案是批處理。

1.簡介

批處理是對批量數(shù)據(jù)執(zhí)行的，無需人工干預(yù)，并且可以長時(shí)間運(yùn)行。 它可能是數(shù)據(jù)或計(jì)算密集型的。 批處理作業(yè)可以按預(yù)定義的時(shí)間表運(yùn)行，也可以按需啟動。 此外，由于批處理作業(yè)通常是長時(shí)間運(yùn)行的作業(yè)，因此在批處理作業(yè)中發(fā)現(xiàn)經(jīng)常檢查和從特定故障中重新啟動是常見的功能。

1.1 Java批處理的歷史

Java平臺的批處理是作為JSR 352規(guī)范（Java EE 7平臺的一部分）引入的，它定義了批處理應(yīng)用程序的編程模型以及用于運(yùn)行和管理批處理作業(yè)的運(yùn)行時(shí)。

1.2 Java Batch的體系結(jié)構(gòu)

下圖顯示了批處理的基本組件。

Java批處理的體系結(jié)構(gòu)

批處理應(yīng)用程序的體系結(jié)構(gòu)解決了批處理問題，例如作業(yè)，步驟，存儲庫，讀取器處理器編寫器模式，塊，檢查點(diǎn)，并行處理，流，重試，排序，分區(qū)等。

讓我們了解架構(gòu)的流程。

• 作業(yè)存儲庫包含需要運(yùn)行的作業(yè)。
• JobLauncher從Job存儲庫中提取一個(gè)作業(yè)。
• 每個(gè)工作都包含步驟。 這些步驟是ItemReader ， ItemProcessor和ItemWriter 。
• Item Reader是讀取數(shù)據(jù)的工具。
• 項(xiàng)目處理是一種將基于業(yè)務(wù)邏輯處理數(shù)據(jù)的處理。
• 條目編寫器會將數(shù)據(jù)寫回到定義的源。

1.3批處理組件。

現(xiàn)在，我們將嘗試詳細(xì)了解批處理組件。

• 作業(yè)：作業(yè)包含整個(gè)批處理過程。 它包含一個(gè)或多個(gè)步驟。 使用作業(yè)指定語言（JSL）將作業(yè)放在一起，該語言指定必須執(zhí)行步驟的順序。 在JSR 352中，JSL在稱為作業(yè)XML文件的XML文件中指定。 一項(xiàng)工作基本上就是一個(gè)存放步驟的容器。
• 步驟：步驟是一個(gè)域?qū)ο?#xff0c;它包含作業(yè)的一個(gè)獨(dú)立的順序階段。 步驟包含執(zhí)行實(shí)際處理所需的所有必要邏輯和數(shù)據(jù)。 根據(jù)批處理規(guī)范，步驟的定義含糊不清，因?yàn)椴襟E的內(nèi)容純粹是特定于應(yīng)用程序的，并且可以像開發(fā)人員所希望的那樣復(fù)雜或簡單。 有兩種步驟： 面向塊和面向任務(wù) 。
• 作業(yè)操作員：它提供了一個(gè)界面來管理作業(yè)處理的各個(gè)方面，其中包括操作命令，例如開始，重新啟動和停止，以及作業(yè)存儲庫命令，例如檢索作業(yè)和步驟執(zhí)行。
• 作業(yè)存儲庫：它包含有關(guān)當(dāng)前正在運(yùn)行的作業(yè)的信息以及有關(guān)該作業(yè)的歷史數(shù)據(jù)。 JobOperator提供用于訪問此存儲庫的API。 JobRepository可以使用數(shù)據(jù)庫或文件系統(tǒng)來實(shí)現(xiàn)。

下一節(jié)將幫助您了解批處理體系結(jié)構(gòu)的一些常見特征。

1.3工作步驟

步驟是作業(yè)的獨(dú)立階段。 如上所述，作業(yè)中有兩種類型的步驟。 我們將在下面嘗試詳細(xì)了解這兩種類型。

1.3.1面向塊的步驟

塊狀步驟將一次讀取和處理一項(xiàng)，并將結(jié)果分組。 然后，當(dāng)塊達(dá)到預(yù)定義的大小時(shí)，將結(jié)果存儲起來。 當(dāng)數(shù)據(jù)集很大時(shí)，面向塊的處理使存儲結(jié)果的效率更高。 它包括三個(gè)部分。

• 項(xiàng)目讀取器從一個(gè)數(shù)據(jù)源中依次讀取輸入，該數(shù)據(jù)源可以是數(shù)據(jù)庫，平面文件，日志文件等。
• 處理器將根據(jù)定義的業(yè)務(wù)邏輯一一處理數(shù)據(jù)。
• 編寫器將數(shù)據(jù)分塊寫入。 塊的大小是預(yù)定義的并且是可配置的

作為塊步驟的一部分，有一些檢查點(diǎn)可為框架提供信息以完成塊。 如果在塊處理期間發(fā)生錯誤，則可以基于最后一個(gè)檢查點(diǎn)重新開始該過程。

1.3.2面向任務(wù)的步驟

除了處理數(shù)據(jù)源中的項(xiàng)目外，它還執(zhí)行任務(wù)。 其中包括創(chuàng)建或刪除目錄，移動文件，創(chuàng)建或刪除數(shù)據(jù)庫表等。與塊步驟相比，任務(wù)步驟通常不會長時(shí)間運(yùn)行。

在正常情況下，在需要清理的面向塊的步驟之后使用面向任務(wù)的步驟。 例如，我們獲取日志文件作為應(yīng)用程序的輸出。 塊步驟用于處理數(shù)據(jù)并從日志文件中獲取有意義的信息。

然后，使用任務(wù)步驟來清理不再需要的較舊的日志文件。

1.3.3并行處理

批處理作業(yè)通常執(zhí)行昂貴的計(jì)算操作并處理大量數(shù)據(jù)。 批處理應(yīng)用程序可以在兩種情況下受益于并行處理。

• 本質(zhì)上獨(dú)立的步驟可以在不同的線程上運(yùn)行。
• 面向塊的步驟（其中每個(gè)項(xiàng)目的處理獨(dú)立于處理先前項(xiàng)目的結(jié)果）可以在多個(gè)線程上運(yùn)行。

批處理有助于完成任務(wù)并更快地執(zhí)行操作以處理海量數(shù)據(jù)。

2.工具與技術(shù)

讓我們看看用于構(gòu)建程序的技術(shù)和工具。

• Eclipse Oxygen.2發(fā)布（4.7.2）
• Java –版本9.0.4
• Gradle– 4.3
• Spring啟動– 2.0.1-發(fā)布
• HSQL數(shù)據(jù)庫

3.項(xiàng)目結(jié)構(gòu)

項(xiàng)目結(jié)構(gòu)如下圖所示。

Java Batch的項(xiàng)目結(jié)構(gòu)

上面的項(xiàng)目結(jié)構(gòu)使用Gradle。 也可以使用maven創(chuàng)建該項(xiàng)目，并且build.gralde將替換為pom.xml文件。 該項(xiàng)目的結(jié)構(gòu)將稍微延遲使用Maven進(jìn)行構(gòu)建。

4.計(jì)劃的目標(biāo)

作為程序的一部分，我們將嘗試使用spring boot創(chuàng)建一個(gè)簡單的Java批處理應(yīng)用程序。 該應(yīng)用程序?qū)?zhí)行以下任務(wù)。

讀取：–從CSV文件讀取員工數(shù)據(jù)。

處理數(shù)據(jù)：–將員工數(shù)據(jù)轉(zhuǎn)換為全部大寫。

寫入：–將已處理的員工數(shù)據(jù)寫回到數(shù)據(jù)庫中。

4.1 Gradle構(gòu)建

我們正在使用Gradle作為程序的一部分進(jìn)行構(gòu)建。 build.gradle文件將如下所示。

build.gradle

buildscript {repositories {mavenCentral()}dependencies {classpath("org.springframework.boot:spring-boot-gradle-plugin:2.0.1.RELEASE")} }apply plugin: 'java' apply plugin: 'eclipse' apply plugin: 'idea' apply plugin: 'org.springframework.boot' apply plugin: 'io.spring.dependency-management'bootJar {baseName = 'java-batch'version = '1.0' }repositories {mavenCentral() }sourceCompatibility = 1.8 targetCompatibility = 1.8dependencies {compile("org.springframework.boot:spring-boot-starter-batch")compile("org.hsqldb:hsqldb")testCompile("junit:junit") }

在上面的build.gradle文件中， apply plugin: 'java'告訴我們需要設(shè)置的插件。 對我們來說，它是Java插件。
repositories{}讓我們知道應(yīng)該從中提取依賴關(guān)系的存儲庫。 我們選擇了mavenCentral拉依賴罐。 我們還可以使用jcenter提取相應(yīng)的依賴罐。

dependencies {}標(biāo)簽用于提供應(yīng)為項(xiàng)目提取的必要的jar文件詳細(xì)信息。 apply plugin: 'org.springframework.boot'此插件用于指定spring-boot項(xiàng)目。 boot jar{}將指定將從構(gòu)建生成的jar的屬性。

4.2樣本數(shù)據(jù)文件

為了提供讀取階段的數(shù)據(jù)，我們將使用包含員工數(shù)據(jù)的CSV文件。

該文件將如下所示。

樣本CSV文件

John,Foster Joe,Toy Justin,Taylor Jane,Clark John,Steve

示例數(shù)據(jù)文件包含員工的名字和姓氏。 我們將使用相同的數(shù)據(jù)進(jìn)行處理，然后將其插入數(shù)據(jù)庫。

4.3 SQL腳本

我們正在使用HSQL數(shù)據(jù)庫，它是基于內(nèi)存的數(shù)據(jù)庫。 該腳本將如下所示。

SQL腳本

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

Spring Boot啟動時(shí)會自動運(yùn)行schema-@@platform@@.sql 。 -all是所有平臺的默認(rèn)設(shè)置。 因此，表創(chuàng)建將在應(yīng)用程序啟動時(shí)自行進(jìn)行，并且在應(yīng)用程序啟動并運(yùn)行之前可用。

4.4模型類別

我們將創(chuàng)建一個(gè)Employee.java類作為模型類。 該類如下所示。

程序的模型類

package com.batch;public class Employee {private String lastName;private String firstName;public Employee() {}public Employee(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;}@Overridepublic String toString() {return "firstName: " + firstName + ", lastName: " + lastName;}}

@Override用于覆蓋toString()方法的默認(rèn)實(shí)現(xiàn)。

4.5配置類

我們將創(chuàng)建一個(gè)BatchConfiguration.java類，它將作為批處理的配置類。 Java文件如下所示。

BatchConfiguration.java

package com.batch.config;import javax.sql.DataSource;import org.springframework.batch.core.Job; import org.springframework.batch.core.JobExecutionListener; import org.springframework.batch.core.Step; import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; import org.springframework.batch.core.configuration.annotation.JobBuilderFactory; import org.springframework.batch.core.configuration.annotation.StepBuilderFactory; import org.springframework.batch.core.launch.support.RunIdIncrementer; import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider; import org.springframework.batch.item.database.JdbcBatchItemWriter; import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder; import org.springframework.batch.item.file.FlatFileItemReader; import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder; import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper; import org.springframework.batch.item.file.mapping.DefaultLineMapper; import org.springframework.batch.item.file.transform.DelimitedLineTokenizer; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.io.ClassPathResource; import org.springframework.jdbc.core.JdbcTemplate;import com.batch.Employee; import com.batch.processor.EmployeeItemProcessor;@Configuration @EnableBatchProcessing public class BatchConfiguration {@Autowiredpublic JobBuilderFactory jobBuilderFactory;@Autowiredpublic StepBuilderFactory stepBuilderFactory;// tag::readerwriterprocessor[]@Beanpublic FlatFileItemReader reader() {return new FlatFileItemReaderBuilder().name("EmployeeItemReader").resource(new ClassPathResource("sample-data.csv")).delimited().names(new String[]{"firstName", "lastName"}).fieldSetMapper(new BeanWrapperFieldSetMapper() {{setTargetType(Employee.class);}}).build();}@Beanpublic EmployeeItemProcessor processor() {return new EmployeeItemProcessor();}@Beanpublic JdbcBatchItemWriter writer(DataSource dataSource) {return new JdbcBatchItemWriterBuilder().itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>()).sql("INSERT INTO employee (first_name, last_name) VALUES (:firstName, :lastName)").dataSource(dataSource).build();}// end::readerwriterprocessor[]// tag::jobstep[]@Beanpublic Job importUserJob(JobCompletionNotificationListener listener, Step step1) {return jobBuilderFactory.get("importUserJob").incrementer(new RunIdIncrementer()).listener(listener).flow(step1).end().build();}@Beanpublic Step step1(JdbcBatchItemWriter writer) {return stepBuilderFactory.get("step1").<Employee, Employee> chunk(10).reader(reader()).processor(processor()).writer(writer).build();}// end::jobstep[] }

@EnableBatchProcessing批注用于啟用批處理。
JobBuilderFactory是用于構(gòu)建作業(yè)的工廠。
StepBuilderFactory用于創(chuàng)建步驟。 方法step1()具有屬性chunk() 。 這是用于將輸入分塊為定義大小的屬性。 對于我們來說，大小為10。

4.6項(xiàng)目處理器

項(xiàng)目處理器是一個(gè)接口，將負(fù)責(zé)處理數(shù)據(jù)。 我們將在EmployeeItemProcessor.java實(shí)現(xiàn)該接口。 Java類如下所示。

EmployeeItemProcessor.java

package com.batch.processor;import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.batch.item.ItemProcessor;import com.batch.Employee;public class EmployeeItemProcessor implements ItemProcessor<Employee, Employee> {private static final Logger log = LoggerFactory.getLogger(EmployeeItemProcessor.class);@Overridepublic Employee process(Employee emp) throws Exception {final String firstName = emp.getFirstName().toUpperCase();final String lastName = emp.getLastName().toUpperCase();final Employee transformedEmployee = new Employee(firstName, lastName);log.info("Converting (" + emp + ") into (" + transformedEmployee + ")");return transformedEmployee;}}

在process()方法中，我們將獲取數(shù)據(jù)，并將其轉(zhuǎn)換為大寫名稱。

4.7 JobExecutionSupportListener類

JobExecutionListenerSupport是將在作業(yè)完成時(shí)通知的接口。 作為接口的一部分，我們有afterJob方法。 此方法用于過帳作業(yè)的完成。

JobCompletionNotificationListener.java

package com.batch.config; import java.util.List;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.jdbc.core.RowMapper; import org.springframework.stereotype.Component;import com.batch.Employee;@Component public class JobCompletionNotificationListener extends JobExecutionListenerSupport {private static final Logger log = LoggerFactory.getLogger(JobCompletionNotificationListener.class);private final JdbcTemplate jdbcTemplate;@Autowiredpublic JobCompletionNotificationListener(JdbcTemplate jdbcTemplate) {this.jdbcTemplate = jdbcTemplate;}@Overridepublic void afterJob(JobExecution jobExecution) {RowMapper rowMapper = (rs, rowNum) -> {Employee e = new Employee();e.setFirstName(rs.getString(1));e.setLastName(rs.getString(2));return e;};if(jobExecution.getStatus() == BatchStatus.COMPLETED) {log.info("!!! JOB FINISHED! Time to verify the results");List empList= jdbcTemplate.query("SELECT first_name, last_name FROM employee",rowMapper);log.info("Size of List "+empList.size());for (Employee emp: empList) {log.info("Found: "+emp.getFirstName()+" "+emp.getLastName());}}} }

在這種方法中，我們是在作業(yè)完成后從數(shù)據(jù)庫中獲取數(shù)據(jù)的，并將結(jié)果打印在控制臺上以驗(yàn)證對數(shù)據(jù)執(zhí)行的處理。

4.8應(yīng)用類別

我們將創(chuàng)建一個(gè)應(yīng)用程序類，其中包含負(fù)責(zé)觸發(fā)Java批處理程序的main方法。 該類如下所示。

應(yīng)用程序

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

@SpringBootApplication是用于將程序指定為Spring Boot程序的注釋。

5.輸出

讓我們將該應(yīng)用程序作為Java應(yīng)用程序執(zhí)行。 我們將在控制臺上獲得以下輸出。

JavaBatch程序的輸出

批處理程序的工作流在輸出中非常清楚。 Job以importUserJob ，然后第一步執(zhí)行開始，將讀取的數(shù)據(jù)轉(zhuǎn)換為大寫。

步驟的后處理，我們可以在控制臺上看到大寫的結(jié)果。

6.總結(jié)

在本教程中，我們學(xué)習(xí)了以下內(nèi)容：

Java批處理包含作業(yè)，該作業(yè)可以包含多個(gè)步驟。

每一步都是閱讀，處理，寫作的結(jié)合。

我們可以將數(shù)據(jù)分塊為不同大小以進(jìn)行處理。

7.下載Eclipse項(xiàng)目

這是帶有SpringBoot的JavaBatch教程。

您可以在此處下載此示例的完整源代碼： JavaBatch.zip

翻譯自: https://www.javacodegeeks.com/2018/05/java-batch-tutorial.html

java批處理 異常處理

總結(jié)

以上是生活随笔為你收集整理的java批处理 异常处理_Java批处理教程的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。