這期內(nèi)容當(dāng)中小編將會給大家?guī)碛嘘P(guān)如何分析Spring Batch遠(yuǎn)程分區(qū)的本地Jar包模式，文章內(nèi)容豐富且以專業(yè)的角度為大家分析和敘述，閱讀完這篇文章希望大家可以有所收獲。
公司主營業(yè)務(wù)：網(wǎng)站設(shè)計、成都網(wǎng)站設(shè)計、移動網(wǎng)站開發(fā)等業(yè)務(wù)。幫助企業(yè)客戶真正實(shí)現(xiàn)互聯(lián)網(wǎng)宣傳，提高企業(yè)的競爭能力。成都創(chuàng)新互聯(lián)公司是一支青春激揚(yáng)、勤奮敬業(yè)、活力青春激揚(yáng)、勤奮敬業(yè)、活力澎湃、和諧高效的團(tuán)隊。公司秉承以“開放、自由、嚴(yán)謹(jǐn)、自律”為核心的企業(yè)文化，感謝他們對我們的高要求，感謝他們從不同領(lǐng)域給我們帶來的挑戰(zhàn)，讓我們激情的團(tuán)隊有機(jī)會用頭腦與智慧不斷的給客戶帶來驚喜。成都創(chuàng)新互聯(lián)公司推出鄂州免費(fèi)做網(wǎng)站回饋大家。
Spring Batch遠(yuǎn)程分區(qū)對于大量數(shù)據(jù)的處理非常擅長，它的實(shí)現(xiàn)有多種方式，如本地Jar包模式、MQ模式、Kubernetes模式。這三種模式的如下：
（1）本地Jar包模式：分區(qū)處理的worker為一個Java進(jìn)程，從jar包啟動，通過jvm參數(shù)和數(shù)據(jù)庫傳遞參數(shù)；官方提供示例代碼。
（2）MQ模式：worker是一個常駐進(jìn)程，Manager和Worker通過消息隊列來傳遞參數(shù)；網(wǎng)上有不少相關(guān)示例代碼。
（3）Kubernetes模式：worker為K8s中的Pod，Manager直接啟動Pod來處理；網(wǎng)上并沒有找到任何示例代碼。
下面將通過代碼來講解第一種模式（本地Jar包模式），其它后續(xù)再介紹。
建議先看下面文章了解一下：
Spring Batch入門：通過例子講解Spring Batch入門，優(yōu)秀的批處理框架
Spring Batch并行處理介紹：大量數(shù)據(jù)也不在話下，Spring Batch并行處理四種模式初探
本文代碼中，Manager和Worker是放在一起的，在同一個項目里，也只會打一個jar包而已；我們通過profile來區(qū)別是manager還是worker，也就是通過Spring Profile實(shí)現(xiàn)一份代碼，兩份邏輯。實(shí)際上也可以拆成兩份代碼，但放一起更方便測試，而且代碼量不大，就沒有必要了。
首先我們需要準(zhǔn)備一個數(shù)據(jù)庫，因?yàn)?code>Manager和Worker都需要同步狀態(tài)到DB上，不能直接使用嵌入式的內(nèi)存數(shù)據(jù)庫了，需要一個外部可共同訪問的數(shù)據(jù)庫。這里我使用的是H2 Database，安裝可參考：把H2數(shù)據(jù)庫從jar包部署到Kubernetes，并解決Ingress不支持TCP的問題。
maven引入依賴如下所示：
org.springframework.boot spring-boot-starter-batch org.springframework.cloud spring-cloud-starter-task com.h3database h3 runtime org.springframework.cloud spring-cloud-deployer-local 2.4.1 org.springframework.batch spring-batch-integration
spring-cloud-deployer-local用于部署和啟動worker，非常關(guān)鍵；其它就是Spring Batch和Task相關(guān)的依賴；以及數(shù)據(jù)庫連接。
Springboot的主類入口如下：
重慶分公司
重慶分公司
@EnableTask
@SpringBootApplication
@EnableBatchProcessing
public class PkslowRemotePartitionJar {
public static void main(String[] args) {
SpringApplication.run(PkslowRemotePartitionJar.class, args);
}
}在Springboot的基礎(chǔ)上，添加了Spring Batch和Spring Cloud Task的支持。
前面的數(shù)據(jù)庫搭建和其它代碼沒有太多可講的，接下來就開始關(guān)鍵代碼的編寫。
Partitioner是遠(yuǎn)程分區(qū)中的核心bean，它定義了分成多少個區(qū)、怎么分區(qū)，要把什么變量傳遞給worker。它會返回一組<分區(qū)名，執(zhí)行上下文>的鍵值對，即返回Map。把要傳遞給worker的變量放在ExecutionContext中去，支持多種類型的變量，如String、int、long等。實(shí)際上，我們不建議通過ExecutionContext來傳遞太多數(shù)據(jù)；可以傳遞一些標(biāo)識或主鍵，然后worker自己去拿數(shù)據(jù)即可。
具體代碼如下：
private static final int GRID_SIZE = 4;
@Bean
public Partitioner partitioner() {
return new Partitioner() {
@Override
public Map partition(int gridSize) {
Map partitions = new HashMap<>(gridSize);
for (int i = 0; i < GRID_SIZE; i++) {
ExecutionContext executionContext = new ExecutionContext();
executionContext.put("partitionNumber", i);
partitions.put("partition" + i, executionContext);
}
return partitions;
}
};
} 上面分成4個區(qū)，程序會啟動4個worker來處理；給worker傳遞的參數(shù)是partitionNumber。
PartitionHandler也是核心的bean，它決定了怎么去啟動worker，給它們傳遞什么jvm參數(shù)（跟之前的ExecutionContext傳遞不一樣）。
@Bean
public PartitionHandler partitionHandler(TaskLauncher taskLauncher, JobExplorer jobExplorer, TaskRepository taskRepository) throws Exception {
Resource resource = this.resourceLoader.getResource(workerResource);
DeployerPartitionHandler partitionHandler =
new DeployerPartitionHandler(taskLauncher, jobExplorer, resource, "workerStep", taskRepository);
List commandLineArgs = new ArrayList<>(3);
commandLineArgs.add("--spring.profiles.active=worker");
commandLineArgs.add("--spring.cloud.task.initialize-enabled=false");
commandLineArgs.add("--spring.batch.initializer.enabled=false");
partitionHandler
.setCommandLineArgsProvider(new PassThroughCommandLineArgsProvider(commandLineArgs));
partitionHandler
.setEnvironmentVariablesProvider(new SimpleEnvironmentVariablesProvider(this.environment));
partitionHandler.setMaxWorkers(2);
partitionHandler.setApplicationName("PkslowWorkerJob");
return partitionHandler;
} 上面代碼中：
resource是worker的jar包地址，表示將啟動該程序；
workerStep是worker將要執(zhí)行的step；
commandLineArgs定義了啟動worker的jvm參數(shù)，如--spring.profiles.active=worker；
environment是manager的系統(tǒng)環(huán)境變量，可以傳遞給worker，當(dāng)然也可以選擇不傳遞；
MaxWorkers是最多能同時啟動多少個worker，類似于線程池大??；設(shè)置為2，表示最多同時有2個worker來處理4個分區(qū)。
完成了分區(qū)相關(guān)的代碼，剩下的就只是如何定義Manager和Worker的業(yè)務(wù)代碼了。
Manager作為管理者，不用太多業(yè)務(wù)邏輯，代碼如下：
@Bean
@Profile("!worker")
public Job partitionedJob(PartitionHandler partitionHandler) throws Exception {
Random random = new Random();
return this.jobBuilderFactory.get("partitionedJob" + random.nextInt())
.start(step1(partitionHandler))
.build();
}
@Bean
public Step step1(PartitionHandler partitionHandler) throws Exception {
return this.stepBuilderFactory.get("step1")
.partitioner(workerStep().getName(), partitioner())
.step(workerStep())
.partitionHandler(partitionHandler)
.build();
}Worker主要作用是處理數(shù)據(jù)，是我們的業(yè)務(wù)代碼，這里就演示一下如何獲取Manager傳遞過來的partitionNumber：
@Bean
public Step workerStep() {
return this.stepBuilderFactory.get("workerStep")
.tasklet(workerTasklet(null, null))
.build();
}
@Bean
@StepScope
public Tasklet workerTasklet(final @Value("#{stepExecutionContext['partitionNumber']}") Integer partitionNumber) {
return new Tasklet() {
@Override
public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
Thread.sleep(6000); //增加延時，查看效果，通過jps：在jar情況下會新起java進(jìn)程
System.out.println("This tasklet ran partition: " + partitionNumber);
return RepeatStatus.FINISHED;
}
};
}通過表達(dá)式@Value("#{stepExecutionContext['partitionNumber']}") 獲取Manager傳遞過來的變量；注意要加注解@StepScope。
因?yàn)槲覀兎譃?code>Manager和Worker，但都是同一份代碼，所以我們先打包一個jar出來，不然manager無法啟動。配置數(shù)據(jù)庫和Worker的jar包地址如下：
spring.datasource.url=jdbc:h3:tcp://localhost:9092/test spring.datasource.username=pkslow spring.datasource.password=pkslow spring.datasource.driver-class-name=org.h3.Driver pkslow.worker.resource=file://pkslow/target/remote-partitioning-jar-1.0-SNAPSHOT.jar
執(zhí)行程序如下：
可以看到啟動了4次Java程序，還給出日志路徑。
通過jps命令查看，能看到一個Manager進(jìn)程，還有兩個worker進(jìn)程：
前面講了Manager可以通過ExecutionContext傳遞變量，如簡單的String、long等。但其實(shí)它也是可以傳遞復(fù)雜的Java對象的，但對應(yīng)的類需要可序列化，如：
import java.io.Serializable;
public class Person implements Serializable {
private Integer age;
private String name;
private String webSite;
//getter and setter
}Manager傳遞：
executionContext.put("person", new Person(0, "pkslow", "www.pkslow.com"));
Worker接收：
@Value("#{stepExecutionContext['person']}") Person person
上述就是小編為大家分享的如何分析Spring Batch遠(yuǎn)程分區(qū)的本地Jar包模式了，如果剛好有類似的疑惑，不妨參照上述分析進(jìn)行理解。如果想知道更多相關(guān)知識，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道。
