1 前言

欢送拜访南瓜慢说 www.pkslow.com获取更多精彩文章！

Spring相干文章：Springboot-Cloud

Spring Batch近程分区对于大量数据的解决十分善于，它的实现有多种形式，如本地Jar包模式、MQ模式、Kubernetes模式。这三种模式的如下：

（1）本地Jar包模式：分区解决的worker为一个Java过程，从jar包启动，通过jvm参数和数据库传递参数；官网提供示例代码。

（2）MQ模式：worker是一个常驻过程，Manager和Worker通过音讯队列来传递参数；网上有不少相干示例代码。

（3）Kubernetes模式：worker为K8s中的Pod，Manager间接启动Pod来解决；网上并没有找到任何示例代码。

本文将通过代码来解说第一种模式（本地Jar包模式），其它后续再介绍。

倡议先看上面文章理解一下：

Spring Batch入门：通过例子解说Spring Batch入门，优良的批处理框架

Spring Batch并行处理介绍：大量数据也不在话下，Spring Batch并行处理四种模式初探

2 代码解说

本文代码中，Manager和Worker是放在一起的，在同一个我的项目里，也只会打一个jar包而已；咱们通过profile来区别是manager还是worker，也就是通过Spring Profile实现一份代码，两份逻辑。实际上也能够拆成两份代码，但放一起更不便测试，而且代码量不大，就没有必要了。

2.1 我的项目筹备

2.1.1 数据库

首先咱们须要筹备一个数据库，因为Manager和Worker都须要同步状态到DB上，不能间接应用嵌入式的内存数据库了，须要一个内部可独特拜访的数据库。这里我应用的是H2 Database，装置可参考：把H2数据库从jar包部署到Kubernetes，并解决Ingress不反对TCP的问题。

2.1.2 引入依赖

maven引入依赖如下所示：

<dependency>  <groupId>org.springframework.boot</groupId>  <artifactId>spring-boot-starter-batch</artifactId></dependency><dependency>  <groupId>org.springframework.cloud</groupId>  <artifactId>spring-cloud-starter-task</artifactId></dependency><dependency>  <groupId>com.h2database</groupId>  <artifactId>h2</artifactId>  <scope>runtime</scope></dependency><dependency>  <groupId>org.springframework.cloud</groupId>  <artifactId>spring-cloud-deployer-local</artifactId>  <version>2.4.1</version></dependency><dependency>  <groupId>org.springframework.batch</groupId>  <artifactId>spring-batch-integration</artifactId></dependency>

spring-cloud-deployer-local用于部署和启动worker，十分要害；其它就是Spring Batch和Task相干的依赖；以及数据库连贯。

2.1.3 主类入口

Springboot的主类入口如下：

@EnableTask@SpringBootApplication@EnableBatchProcessingpublic class PkslowRemotePartitionJar {    public static void main(String[] args) {        SpringApplication.run(PkslowRemotePartitionJar.class, args);    }}

在Springboot的根底上，增加了Spring Batch和Spring Cloud Task的反对。

2.2 要害代码编写

后面的数据库搭建和其它代码没有太多可讲的，接下来就开始要害代码的编写。

2.2.1 分区治理Partitioner

Partitioner是近程分区中的外围bean，它定义了分成多少个区、怎么分区，要把什么变量传递给worker。它会返回一组<分区名，执行上下文>的键值对，即返回Map<String, ExecutionContext>。把要传递给worker的变量放在ExecutionContext中去，反对多种类型的变量，如String、int、long等。实际上，咱们不倡议通过ExecutionContext来传递太多数据；能够传递一些标识或主键，而后worker本人去拿数据即可。

具体代码如下：

private static final int GRID_SIZE = 4;@Beanpublic Partitioner partitioner() {  return new Partitioner() {    @Override    public Map<String, ExecutionContext> partition(int gridSize) {      Map<String, ExecutionContext> 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个区，程序会启动4个worker来解决；给worker传递的参数是partitionNumber。

2.2.2 分区处理器PartitionHandler

PartitionHandler也是外围的bean，它决定了怎么去启动worker，给它们传递什么jvm参数（跟之前的ExecutionContext传递不一样）。

@Beanpublic 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<String> 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将要执行的step；

commandLineArgs定义了启动worker的jvm参数，如--spring.profiles.active=worker；

environment是manager的零碎环境变量，能够传递给worker，当然也能够抉择不传递；

MaxWorkers是最多能同时启动多少个worker，相似于线程池大小；设置为2，示意最多同时有2个worker来解决4个分区。

2.2.3 Manager和Worker的Batch定义

实现了分区相干的代码，剩下的就只是如何定义Manager和Worker的业务代码了。

Manager作为管理者，不必太多业务逻辑，代码如下：

@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();}@Beanpublic Step step1(PartitionHandler partitionHandler) throws Exception {  return this.stepBuilderFactory.get("step1")    .partitioner(workerStep().getName(), partitioner())    .step(workerStep())    .partitionHandler(partitionHandler)    .build();}

Worker次要作用是解决数据，是咱们的业务代码，这里就演示一下如何获取Manager传递过去的partitionNumber：

@Beanpublic Step workerStep() {  return this.stepBuilderFactory.get("workerStep")    .tasklet(workerTasklet(null, null))    .build();}@Bean@StepScopepublic 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过程      System.out.println("This tasklet ran partition: " + partitionNumber);           return RepeatStatus.FINISHED;    }  };}

通过表达式@Value("#{stepExecutionContext['partitionNumber']}") 获取Manager传递过去的变量；留神要加注解@StepScope。

3 程序运行

因为咱们分为Manager和Worker，但都是同一份代码，所以咱们先打包一个jar进去，不然manager无奈启动。配置数据库和Worker的jar包地址如下：

spring.datasource.url=jdbc:h2:tcp://localhost:9092/testspring.datasource.username=pkslowspring.datasource.password=pkslowspring.datasource.driver-class-name=org.h2.Driverpkslow.worker.resource=file://pkslow/target/remote-partitioning-jar-1.0-SNAPSHOT.jar

执行程序如下：

能够看到启动了4次Java程序，还给出日志门路。

通过jps命令查看，能看到一个Manager过程，还有两个worker过程：

4 简单变量传递

后面讲了Manager能够通过ExecutionContext传递变量，如简略的String、long等。但其实它也是能够传递简单的Java对象的，但对应的类须要可序列化，如：

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

5 总结

本文介绍了Spring Batch近程分区的本地Jar包模式，只能在一台机器上运行，所以也是无奈真正施展出近程分区的作用。但它对咱们后续了解更简单的模式是有很大帮忙的；同时，咱们也能够应用本地模式进行开发测试，毕竟它只须要一个数据库就行了，依赖很少。

欢送关注微信公众号<南瓜慢说>，将继续为你更新...

多读书，多分享；多写作，多整顿。