Spring Batch远程分区的本地Jar包模式的代码详解
1前言
SpringBatch远程分区对于大量数据的处理非常擅长,它的实现有多种方式,如本地Jar包模式、MQ模式、Kubernetes模式。这三种模式的如下:
(1)本地Jar包模式:分区处理的worker为一个Java进程,从jar包启动,通过jvm参数和数据库传递参数;官方提供示例代码。
(2)MQ模式:worker是一个常驻进程,Manager和Worker通过消息队列来传递参数;网上有不少相关示例代码。
(3)Kubernetes模式:worker为K8s中的Pod,Manager直接启动Pod来处理;网上并没有找到任何示例代码。
本文将通过代码来讲解第一种模式(本地Jar包模式),其它后续再介绍。
建议先看下面文章了解一下:
SpringBatch入门:SpringBatch入门教程篇
SpringBatch并行处理介绍:详解SpringBoot和SpringBatch使用
2代码讲解
本文代码中,Manager和Worker是放在一起的,在同一个项目里,也只会打一个jar包而已;我们通过profile来区别是manager还是worker,也就是通过SpringProfile实现一份代码,两份逻辑。实际上也可以拆成两份代码,但放一起更方便测试,而且代码量不大,就没有必要了。
2.1项目准备
2.1.1数据库
首先我们需要准备一个数据库,因为Manager和Worker都需要同步状态到DB上,不能直接使用嵌入式的内存数据库了,需要一个外部可共同访问的数据库。这里我使用的是H2Database,安装可参考:把H2数据库从jar包部署到Kubernetes,并解决Ingress不支持TCP的问题。
2.1.2引入依赖
maven引入依赖如下所示:
org.springframework.boot spring-boot-starter-batch org.springframework.cloud spring-cloud-starter-task com.h2database h2 runtime org.springframework.cloud spring-cloud-deployer-local 2.4.1 org.springframework.batch spring-batch-integration
spring-cloud-deployer-local用于部署和启动worker,非常关键;其它就是SpringBatch和Task相关的依赖;以及数据库连接。
2.1.3主类入口
Springboot的主类入口如下:
@EnableTask @SpringBootApplication @EnableBatchProcessing publicclassPkslowRemotePartitionJar{ publicstaticvoidmain(String[]args){ SpringApplication.run(PkslowRemotePartitionJar.class,args); } }
在Springboot的基础上,添加了SpringBatch和SpringCloudTask的支持。
2.2关键代码编写
前面的数据库搭建和其它代码没有太多可讲的,接下来就开始关键代码的编写。
2.2.1分区管理Partitioner
Partitioner是远程分区中的核心bean,它定义了分成多少个区、怎么分区,要把什么变量传递给worker。它会返回一组<分区名,执行上下文>的键值对,即返回Map
具体代码如下:
privatestaticfinalintGRID_SIZE=4; @Bean publicPartitionerpartitioner(){ returnnewPartitioner(){ @Override publicMappartition(intgridSize){ Map partitions=newHashMap<>(gridSize); for(inti=0;i 上面分成4个区,程序会启动4个worker来处理;给worker传递的参数是partitionNumber。
2.2.2分区处理器PartitionHandler
PartitionHandler也是核心的bean,它决定了怎么去启动worker,给它们传递什么jvm参数(跟之前的ExecutionContext传递不一样)。
@Bean publicPartitionHandlerpartitionHandler(TaskLaunchertaskLauncher,JobExplorerjobExplorer,TaskRepositorytaskRepository)throwsException{ Resourceresource=this.resourceLoader.getResource(workerResource); DeployerPartitionHandlerpartitionHandler= newDeployerPartitionHandler(taskLauncher,jobExplorer,resource,"workerStep",taskRepository); ListcommandLineArgs=newArrayList<>(3); commandLineArgs.add("--spring.profiles.active=worker"); commandLineArgs.add("--spring.cloud.task.initialize-enabled=false"); commandLineArgs.add("--spring.batch.initializer.enabled=false"); partitionHandler .setCommandLineArgsProvider(newPassThroughCommandLineArgsProvider(commandLineArgs)); partitionHandler .setEnvironmentVariablesProvider(newSimpleEnvironmentVariablesProvider(this.environment)); partitionHandler.setMaxWorkers(2); partitionHandler.setApplicationName("PkslowWorkerJob"); returnpartitionHandler; } 上面代码中:
resource是worker的jar包地址,表示将启动该程序;
workerStep是worker将要执行的step;
commandLineArgs定义了启动worker的jvm参数,如--spring.profiles.active=worker;
environment是manager的系统环境变量,可以传递给worker,当然也可以选择不传递;
MaxWorkers是最多能同时启动多少个worker,类似于线程池大小;设置为2,表示最多同时有2个worker来处理4个分区。
2.2.3Manager和Worker的Batch定义
完成了分区相关的代码,剩下的就只是如何定义Manager和Worker的业务代码了。
Manager作为管理者,不用太多业务逻辑,代码如下:
@Bean @Profile("!worker") publicJobpartitionedJob(PartitionHandlerpartitionHandler)throwsException{ Randomrandom=newRandom(); returnthis.jobBuilderFactory.get("partitionedJob"+random.nextInt()) .start(step1(partitionHandler)) .build(); } @Bean publicStepstep1(PartitionHandlerpartitionHandler)throwsException{ returnthis.stepBuilderFactory.get("step1") .partitioner(workerStep().getName(),partitioner()) .step(workerStep()) .partitionHandler(partitionHandler) .build(); }Worker主要作用是处理数据,是我们的业务代码,这里就演示一下如何获取Manager传递过来的partitionNumber:
@Bean publicStepworkerStep(){ returnthis.stepBuilderFactory.get("workerStep") .tasklet(workerTasklet(null,null)) .build(); } @Bean @StepScope publicTaskletworkerTasklet(final@Value("#{stepExecutionContext['partitionNumber']}")IntegerpartitionNumber){ returnnewTasklet(){ @Override publicRepeatStatusexecute(StepContributioncontribution,ChunkContextchunkContext)throwsException{ Thread.sleep(6000);//增加延时,查看效果,通过jps:在jar情况下会新起java进程 System.out.println("Thistaskletranpartition:"+partitionNumber); returnRepeatStatus.FINISHED; } }; }通过表达式@Value("#{stepExecutionContext['partitionNumber']}")获取Manager传递过来的变量;注意要加注解@StepScope。
3程序运行
因为我们分为Manager和Worker,但都是同一份代码,所以我们先打包一个jar出来,不然manager无法启动。配置数据库和Worker的jar包地址如下:
spring.datasource.url=jdbc:h2:tcp://localhost:9092/test spring.datasource.username=pkslow spring.datasource.password=pkslow spring.datasource.driver-class-name=org.h2.Driver pkslow.worker.resource=file://pkslow/target/remote-partitioning-jar-1.0-SNAPSHOT.jar执行程序如下:
可以看到启动了4次Java程序,还给出日志路径。
通过jps命令查看,能看到一个Manager进程,还有两个worker进程:
4复杂变量传递
前面讲了Manager可以通过ExecutionContext传递变量,如简单的String、long等。但其实它也是可以传递复杂的Java对象的,但对应的类需要可序列化,如:
importjava.io.Serializable; publicclassPersonimplementsSerializable{ privateIntegerage; privateStringname; privateStringwebSite; //getterandsetter }Manager传递:
executionContext.put("person",newPerson(0,"pkslow","www.pkslow.com"));
Worker接收:
@Value("#{stepExecutionContext['person']}")Personperson
5总结
本文介绍了SpringBatch远程分区的本地Jar包模式,只能在一台机器上运行,所以也是无法真正发挥出远程分区的作用。但它对我们后续理解更复杂的模式是有很大帮助的;同时,我们也可以使用本地模式进行开发测试,毕竟它只需要一个数据库就行了,依赖很少。
声明:本文内容来源于网络,版权归原作者所有,内容由互联网用户自发贡献自行上传,本网站不拥有所有权,未作人工编辑处理,也不承担相关法律责任。如果您发现有涉嫌版权的内容,欢迎发送邮件至:czq8825#qq.com(发邮件时,请将#更换为@)进行举报,并提供相关证据,一经查实,本站将立刻删除涉嫌侵权内容。