Spring 全家桶中的批處理框架，真心強大！ - 網安 - 專業的網絡安全產業、社區、知識平臺

spring batch簡介

spring batch是spring提供的一個數據處理框架。企業域中的許多應用程序需要批量處理才能在關鍵任務環境中執行業務操作。這些業務運營包括：

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

Spring Batch是一個輕量級，全面的批處理框架，旨在開發對企業系統日常運營至關重要的強大批處理應用程序。Spring Batch構建了人們期望的Spring Framework特性（生產力，基于POJO的開發方法和一般易用性），同時使開發人員可以在必要時輕松訪問和利用更高級的企業服務。Spring Batch不是一個schuedling的框架。

Spring Batch提供了可重用的功能，這些功能對于處理大量的數據至關重要，包括記錄/跟蹤，事務管理，作業處理統計，作業重啟，跳過和資源管理。它還提供更高級的技術服務和功能，通過優化和分區技術實現極高容量和高性能的批處理作業。

Spring Batch可用于兩種簡單的用例（例如將文件讀入數據庫或運行存儲過程）以及復雜的大量用例（例如在數據庫之間移動大量數據，轉換它等等）上）。大批量批處理作業可以高度可擴展的方式利用該框架來處理大量信息。

- Spring Batch架構介紹

一個典型的批處理應用程序大致如下：

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

其對應的示意圖如下：

spring batch的一個總體的架構如下：

在spring batch中一個job可以定義很多的步驟step，在每一個step里面可以定義其專屬的ItemReader用于讀取數據，ItemProcesseor用于處理數據，ItemWriter用于寫數據，而每一個定義的job則都在JobRepository里面，我們可以通過JobLauncher來啟動某一個job。

Spring Batch核心概念介紹

下面是一些概念是Spring batch框架中的核心概念。

什么是Job

Job和Step是spring batch執行批處理任務最為核心的兩個概念。

其中Job是一個封裝整個批處理過程的一個概念。Job在spring batch的體系當中只是一個最頂層的一個抽象概念，體現在代碼當中則它只是一個最上層的接口，其代碼如下:

/**
 * Batch domain object representing a job. Job is an explicit abstraction
 * representing the configuration of a job specified by a developer. It should
 * be noted that restart policy is applied to the job as a whole and not to a
 * step.
 */
public interface Job {
 
 String getName();
 
 
 boolean isRestartable();
 
 
 void execute(JobExecution execution);
 
 
 JobParametersIncrementer getJobParametersIncrementer();
 
 
 JobParametersValidator getJobParametersValidator();
 
}

在Job這個接口當中定義了五個方法，它的實現類主要有兩種類型的job，一個是simplejob，另一個是flowjob。在spring batch當中，job是最頂層的抽象，除job之外我們還有JobInstance以及JobExecution這兩個更加底層的抽象。

一個job是我們運行的基本單位，它內部由step組成。job本質上可以看成step的一個容器。一個job可以按照指定的邏輯順序組合step，并提供了我們給所有step設置相同屬性的方法，例如一些事件監聽，跳過策略。

Spring Batch以SimpleJob類的形式提供了Job接口的默認簡單實現，它在Job之上創建了一些標準功能。一個使用java config的例子代碼如下：

@Bean
public Job footballJob() {
    return this.jobBuilderFactory.get("footballJob")
                     .start(playerLoad())
                     .next(gameLoad())
                     .next(playerSummarization())
                     .end()
                     .build();
}

這個配置的意思是：首先給這個job起了一個名字叫footballJob，接著指定了這個job的三個step，他們分別由方法，playerLoad,gameLoad, playerSummarization實現。

什么是JobInstance

我們在上文已經提到了JobInstance，他是Job的更加底層的一個抽象，他的定義如下：

public interface JobInstance {
 /**
  * Get unique id for this JobInstance.
  * @return instance id
  */
 public long getInstanceId();
 /**
  * Get job name.
  * @return value of 'id' attribute from 
  */
 public String getJobName(); 
}

他的方法很簡單，一個是返回Job的id，另一個是返回Job的名字。

JobInstance指的是job運行當中，作業執行過程當中的概念。Instance本就是實例的意思。

比如說現在有一個批處理的job，它的功能是在一天結束時執行行一次。我們假定這個批處理job的名字為'EndOfDay'。在這個情況下，那么每天就會有一個邏輯意義上的JobInstance, 而我們必須記錄job的每次運行的情況。

什么是JobParameters

在上文當中我們提到了，同一個job每天運行一次的話，那么每天都有一個jobIntsance，但他們的job定義都是一樣的，那么我們怎么來區別一個job的不同jobinstance了。不妨先做個猜想，雖然jobinstance的job定義一樣，但是他們有的東西就不一樣，例如運行時間。

spring batch中提供的用來標識一個jobinstance的東西是：JobParameters。JobParameters對象包含一組用于啟動批處理作業的參數，它可以在運行期間用于識別或甚至用作參考數據。我們假設的運行時間，就可以作為一個JobParameters。

例如, 我們前面的'EndOfDay'的job現在已經有了兩個實例，一個產生于1月1日，另一個產生于1月2日，那么我們就可以定義兩個JobParameter對象：一個的參數是01-01, 另一個的參數是01-02。因此，識別一個JobInstance的方法可以定義為：

因此，我么可以通過Jobparameter來操作正確的JobInstance

什么是JobExecution

JobExecution指的是單次嘗試運行一個我們定義好的Job的代碼層面的概念。job的一次執行可能以失敗也可能成功。只有當執行成功完成時，給定的與執行相對應的JobInstance才也被視為完成。

還是以前面描述的EndOfDay的job作為示例，假設第一次運行01-01-2019的JobInstance結果是失敗。那么此時如果使用與第一次運行相同的Jobparameter參數（即01-01-2019）作業參數再次運行，那么就會創建一個對應于之前jobInstance的一個新的JobExecution實例,JobInstance仍然只有一個。

JobExecution的接口定義如下：

public interface JobExecution {
 /**
  * Get unique id for this JobExecution.
  * @return execution id
  */
 public long getExecutionId();
 /**
  * Get job name.
  * @return value of 'id' attribute from 
  */
 public String getJobName(); 
 /**
  * Get batch status of this execution.
  * @return batch status value.
  */
 public BatchStatus getBatchStatus();
 /**
  * Get time execution entered STARTED status. 
  * @return date (time)
  */
 public Date getStartTime();
 /**
  * Get time execution entered end status: COMPLETED, STOPPED, FAILED 
  * @return date (time)
  */
 public Date getEndTime();
 /**
  * Get execution exit status.
  * @return exit status.
  */
 public String getExitStatus();
 /**
  * Get time execution was created.
  * @return date (time)
  */
 public Date getCreateTime();
 /**
  * Get time execution was last updated updated.
  * @return date (time)
  */
 public Date getLastUpdatedTime();
 /**
  * Get job parameters for this execution.
  * @return job parameters  
  */
 public Properties getJobParameters();
 
}

每一個方法的注釋已經解釋的很清楚，這里不再多做解釋。只提一下BatchStatus，JobExecution當中提供了一個方法getBatchStatus用于獲取一個job某一次特地執行的一個狀態。BatchStatus是一個代表job狀態的枚舉類，其定義如下：

public enum BatchStatus {STARTING, STARTED, STOPPING, 
   STOPPED, FAILED, COMPLETED, ABANDONED }

這些屬性對于一個job的執行來說是非常關鍵的信息，并且spring batch會將他們持久到數據庫當中. 在使用Spring batch的過程當中spring batch會自動創建一些表用于存儲一些job相關的信息，用于存儲JobExecution的表為batch_job_execution,下面是一個從數據庫當中截圖的實例：

什么是Step

每一個Step對象都封裝了批處理作業的一個獨立的階段。事實上，每一個Job本質上都是由一個或多個步驟組成。每一個step包含定義和控制實際批處理所需的所有信息。

任何特定的內容都由編寫Job的開發人員自行決定。一個step可以非常簡單也可以非常復雜。例如，一個step的功能是將文件中的數據加載到數據庫中，那么基于現在spring batch的支持則幾乎不需要寫代碼。更復雜的step可能具有復雜的業務邏輯，這些邏輯作為處理的一部分。

與Job一樣，Step具有與JobExecution類似的StepExecution，如下圖所示：

什么是StepExecution

StepExecution表示一次執行Step, 每次運行一個Step時都會創建一個新的StepExecution，類似于JobExecution。但是，某個步驟可能由于其之前的步驟失敗而無法執行。且僅當Step實際啟動時才會創建StepExecution。

一次step執行的實例由StepExecution類的對象表示。每個StepExecution都包含對其相應步驟的引用以及JobExecution和事務相關的數據，例如提交和回滾計數以及開始和結束時間。

此外，每個步驟執行都包含一個ExecutionContext，其中包含開發人員需要在批處理運行中保留的任何數據，例如重新啟動所需的統計信息或狀態信息。下面是一個從數據庫當中截圖的實例：

什么是ExecutionContext

ExecutionContext`即每一個`StepExecution` 的執行環境。它包含一系列的鍵值對。我們可以用如下代碼獲取`ExecutionContext
ExecutionContext ecStep = stepExecution.getExecutionContext();
ExecutionContext ecJob = jobExecution.getExecutionContext();

什么是JobRepository

JobRepository是一個用于將上述job，step等概念進行持久化的一個類。它同時給Job和Step以及下文會提到的JobLauncher實現提供CRUD操作。

首次啟動Job時，將從repository中獲取JobExecution，并且在執行批處理的過程中，StepExecution和JobExecution將被存儲到repository當中。

@EnableBatchProcessing注解可以為JobRepository提供自動配置。

什么是JobLauncher

JobLauncher這個接口的功能非常簡單，它是用于啟動指定了JobParameters的Job，為什么這里要強調指定了JobParameter，原因其實我們在前面已經提到了，jobparameter和job一起才能組成一次job的執行。下面是代碼實例：

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

上面run方法實現的功能是根據傳入的job以及jobparamaters從JobRepository獲取一個JobExecution并執行Job。

什么是Item Reader

ItemReader是一個讀數據的抽象，它的功能是為每一個Step提供數據輸入。當ItemReader以及讀完所有數據時，它會返回null來告訴后續操作數據已經讀完。Spring Batch為ItemReader提供了非常多的有用的實現類，比如JdbcPagingItemReader，JdbcCursorItemReader等等。

ItemReader支持的讀入的數據源也是非常豐富的，包括各種類型的數據庫，文件，數據流，等等。幾乎涵蓋了我們的所有場景。

下面是一個JdbcPagingItemReader的例子代碼：

@Bean
public JdbcPagingItemReader itemReader(DataSource dataSource, PagingQueryProvider queryProvider) {
        Map parameterValues = new HashMap<>();
        parameterValues.put("status", "NEW");
 
        return new JdbcPagingItemReaderBuilder()
               .name("creditReader")
               .dataSource(dataSource)
               .queryProvider(queryProvider)
               .parameterValues(parameterValues)
               .rowMapper(customerCreditMapper())
               .pageSize(1000)
               .build();
}
 
@Bean
public SqlPagingQueryProviderFactoryBean queryProvider() {
        SqlPagingQueryProviderFactoryBean provider = new SqlPagingQueryProviderFactoryBean();
 
        provider.setSelectClause("select id, name, credit");
        provider.setFromClause("from customer");
        provider.setWhereClause("where status=:status");
        provider.setSortKey("id");
 
        return provider;
}

JdbcPagingItemReader必須指定一個PagingQueryProvider，負責提供SQL查詢語句來按分頁返回數據。

下面是一個JdbcCursorItemReader的例子代碼:

 private JdbcCursorItemReader> buildItemReader(final DataSource dataSource, String tableName,
            String tenant) {
 
        JdbcCursorItemReader> itemReader = new JdbcCursorItemReader<>();
        itemReader.setDataSource(dataSource);
        itemReader.setSql("sql here");
        itemReader.setRowMapper(new RowMapper());
        return itemReader;
    }

什么是Item Writer

既然ItemReader是讀數據的一個抽象，那么ItemWriter自然就是一個寫數據的抽象，它是為每一個step提供數據寫出的功能。寫的單位是可以配置的，我們可以一次寫一條數據，也可以一次寫一個chunk的數據，關于chunk下文會有專門的介紹。ItemWriter對于讀入的數據是不能做任何操作的。

Spring Batch為ItemWriter也提供了非常多的有用的實現類，當然我們也可以去實現自己的writer功能。

什么是Item Processor

ItemProcessor對項目的業務邏輯處理的一個抽象, 當ItemReader讀取到一條記錄之后，ItemWriter還未寫入這條記錄之前，I我們可以借助temProcessor提供一個處理業務邏輯的功能，并對數據進行相應操作。

如果我們在ItemProcessor發現一條數據不應該被寫入，可以通過返回null來表示。ItemProcessor和ItemReader以及ItemWriter可以非常好的結合在一起工作，他們之間的數據傳輸也非常方便。我們直接使用即可。

chunk 處理流程

spring batch提供了讓我們按照chunk處理數據的能力，一個chunk的示意圖如下：

它的意思就和圖示的一樣，由于我們一次batch的任務可能會有很多的數據讀寫操作，因此一條一條的處理并向數據庫提交的話效率不會很高，因此spring batch提供了chunk這個概念，我們可以設定一個chunk size，spring batch 將一條一條處理數據，但不提交到數據庫，只有當處理的數據數量達到chunk size設定的值得時候，才一起去commit.

java的實例定義代碼如下：

在上面這個step里面，chunk size被設為了10，當ItemReader讀的數據數量達到10的時候，這一批次的數據就一起被傳到itemWriter，同時transaction被提交。

skip策略和失敗處理

一個batch的job的step，可能會處理非常大數量的數據，難免會遇到出錯的情況，出錯的情況雖出現的概率較小，但是我們不得不考慮這些情況，因為我們做數據遷移最重要的是要保證數據的最終一致性。spring batch當然也考慮到了這種情況，并且為我們提供了相關的技術支持，請看如下bean的配置：

我們需要留意這三個方法，分別是skipLimit(),skip(),noSkip(),

skipLimit方法的意思是我們可以設定一個我們允許的這個step可以跳過的異常數量，假如我們設定為10，則當這個step運行時，只要出現的異常數目不超過10，整個step都不會fail。

注意，若不設定skipLimit，則其默認值是0.

skip方法我們可以指定我們可以跳過的異常，因為有些異常的出現，我們是可以忽略的。

noSkip方法的意思則是指出現這個異常我們不想跳過，也就是從skip的所以exception當中排除這個exception，從上面的例子來說，也就是跳過所有除FileNotFoundException的exception。那么對于這個step來說，FileNotFoundException就是一個fatal的exception，拋出這個exception的時候step就會直接fail

批處理操作指南

本部分是一些使用spring batch時的值得注意的點

- 批處理原則

在構建批處理解決方案時，應考慮以下關鍵原則和注意事項。

批處理體系結構通常會影響體系結構
盡可能簡化并避免在單批應用程序中構建復雜的邏輯結構
保持數據的處理和存儲在物理上靠得很近（換句話說，將數據保存在處理過程中）。
最大限度地減少系統資源的使用，尤其是I / O. 在internal memory中執行盡可能多的操作。
查看應用程序I / O（分析SQL語句）以確保避免不必要的物理I / O. 特別是，需要尋找以下四個常見缺陷：
當數據可以被讀取一次并緩存或保存在工作存儲中時，讀取每個事務的數據。
重新讀取先前在同一事務中讀取數據的事務的數據。
導致不必要的表或索引掃描。
未在SQL語句的WHERE子句中指定鍵值。
在批處理運行中不要做兩次一樣的事情。例如，如果需要數據匯總以用于報告目的，則應該（如果可能）在最初處理數據時遞增存儲的總計，因此您的報告應用程序不必重新處理相同的數據。
在批處理應用程序開始時分配足夠的內存，以避免在此過程中進行耗時的重新分配。
總是假設數據完整性最差。插入適當的檢查和記錄驗證以維護數據完整性。
盡可能實施校驗和以進行內部驗證。例如，對于一個文件里的數據應該有一個數據條數紀錄，告訴文件中的記錄總數以及關鍵字段的匯總。
在具有真實數據量的類似生產環境中盡早計劃和執行壓力測試。
在大批量系統中，數據備份可能具有挑戰性，特別是如果系統以24-7在線的情況運行。數據庫備份通常在在線設計中得到很好的處理，但文件備份應該被視為同樣重要。如果系統依賴于文件，則文件備份過程不僅應該到位并記錄在案，還應定期進行測試。

如何默認不啟動job

在使用java config使用spring batch的job時，如果不做任何配置，項目在啟動時就會默認去跑我們定義好的批處理job。那么如何讓項目在啟動時不自動去跑job呢？

spring batch的job會在項目啟動時自動run，如果我們不想讓他在啟動時run的話，可以在application.properties中添加如下屬性：

spring.batch.job.enabled=false

在讀數據時內存不夠

在使用spring batch做數據遷移時，發現在job啟動后，執行到一定時間點時就卡在一個地方不動了，且log也不再打印，等待一段時間之后，得到如下錯誤：

紅字的信息為：Resource exhaustion event：the JVM was unable to allocate memory from the heap.

翻譯過來的意思就是項目發出了一個資源耗盡的事件，告訴我們java虛擬機無法再為堆分配內存。

造成這個錯誤的原因是: 這個項目里的batch job的reader是一次性拿回了數據庫里的所有數據，并沒有進行分頁，當這個數據量太大時，就會導致內存不夠用。解決的辦法有兩個:

調整reader讀數據邏輯，按分頁讀取，但實現上會麻煩一些，且運行效率會下降
增大service內存