SpringBatch 系列之 Remote-Chunking

用户头像
稻草鸟人
关注
发布于: 2020 年 06 月 28 日
SpringBatch系列之Remote-Chunking

1、概要



前面的文章介绍了Spring Batch并发并行的批处理能力,但是还不够,单台机器的性能终归有极限,因此我们有些场景就可以考虑使用多台机器来处理。



本文我们将介绍remote chunking,第一篇简单介绍Spring Batch多机器处理披露任务的能力。



2、什么是`remote chunking`



remote chunking将数据读和写拆分到一个master多个slave机器上。master机器负责读数据并且分发数据到slave机器。master机器在Step中读取数据,并通过像JMS这样的技术将块处理部分交给salve机器。





在master端,RemoteChunkingManagerStepBuilderFactory允许我们通过声明如下内容配置master步骤



  • 配置item reader读取数据发送给workers

  • 配置output channel(Outgoing requests)发送请求给workers

  • 配置input channel(Incoming replies)接收workers响应



没有必要显式声明ChunkMessageChannelItemWriterMessagingTemplate默认即可(如果需要也可以以显式配置出来)



在worker端,RemoteChunkingWorkerBuilder允许有如下配置



  • 通过input channel(Incoming requests)监听master端发出的请求

  • 为每一个请求调用ChunkProcessorChunkHandlerhandleChunk方法执行配置好的ItemProcessorItemWriter

  • 通过output channel (Outgoing replies)发送响应到master端



没有必要显式声明SimpleChunkProcessorChunkProcessorChunkHandler默认即可(如有必要也可以显式配置出来)



从4.1版本开始,Spring Batch Integration通过注解@EnableBatchIntegration简化了remote chunking步骤。这个注解主要作用是方便注入如下两个bean



  • RemoteChunkingManagerStepBuilderFactory: 在master端配置

  • RemoteChunkingWorkerBuilder:用于配置worker端处理流程



3、开始多进程之旅



3.1、添加依赖



在之前的maven配置基础之上,添加如下依赖



<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.batch</groupId>
<artifactId>spring-batch-integration</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-activemq</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-jms</artifactId>
</dependency>



3.2、配置ActiveMQ



broker.url=tcp://localhost:61616



3.3、Master端程序



@Profile("master")
public class ManagerConfiguration {
@Value("${broker.url}")
private String brokerUrl;
@Autowired
private JobBuilderFactory jobBuilderFactory;
@Autowired
private RemoteChunkingManagerStepBuilderFactory managerStepBuilderFactory;
@Bean
public ActiveMQConnectionFactory connectionFactory() {
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
connectionFactory.setBrokerURL(this.brokerUrl);
connectionFactory.setTrustAllPackages(true);
return connectionFactory;
}
/*
* Configure outbound flow (requests going to workers)
*/
@Bean
public DirectChannel requests() {
return new DirectChannel();
}
@Bean
public IntegrationFlow outboundFlow(ActiveMQConnectionFactory connectionFactory) {
return IntegrationFlows
.from(requests())
.handle(Jms.outboundAdapter(connectionFactory).destination("requests"))
.get();
}
/*
* Configure inbound flow (replies coming from workers)
*/
@Bean
public QueueChannel replies() {
return new QueueChannel();
}
@Bean
public IntegrationFlow inboundFlow(ActiveMQConnectionFactory connectionFactory) {
return IntegrationFlows
.from(Jms.messageDrivenChannelAdapter(connectionFactory).destination("replies"))
.channel(replies())
.get();
}
/*
* Configure master step components
*/
@Bean
public ListItemReader<Integer> itemReader() {
return new ListItemReader<>(Arrays.asList(1, 2, 3, 4, 5, 6));
}
@Bean
public TaskletStep managerStep() {
return this.managerStepBuilderFactory.get("managerStep")
.<Integer, Integer>chunk(3)
.reader(itemReader())
.outputChannel(requests())
.inputChannel(replies())
.build();
}
@Bean("remoteChunkingJob")
public Job remoteChunkingJob() {
return this.jobBuilderFactory.get("remoteChunkingJob")
.start(managerStep())
.build();
}
}



3.4、Worker端程序



@Profile("worker")
public class WorkerConfiguration {
@Value("${broker.url}")
private String brokerUrl;
@Resource
private RemoteChunkingWorkerBuilder<Integer, Integer> remoteChunkingWorkerBuilder;
@Bean
public ActiveMQConnectionFactory connectionFactory() {
ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
connectionFactory.setBrokerURL(this.brokerUrl);
connectionFactory.setTrustAllPackages(true);
return connectionFactory;
}
/*
* Configure inbound flow (requests coming from the master)
*/
@Bean
public DirectChannel requests() {
return new DirectChannel();
}
@Bean
public IntegrationFlow inboundFlow(ActiveMQConnectionFactory connectionFactory) {
return IntegrationFlows
.from(Jms.messageDrivenChannelAdapter(connectionFactory).destination("requests"))
.channel(requests())
.get();
}
/*
* Configure outbound flow (replies going to the master)
*/
@Bean
public DirectChannel replies() {
return new DirectChannel();
}
@Bean
public IntegrationFlow outboundFlow(ActiveMQConnectionFactory connectionFactory) {
return IntegrationFlows
.from(replies())
.handle(Jms.outboundAdapter(connectionFactory).destination("replies"))
.get();
}
/*
* Configure worker components
*/
@Bean
public ItemProcessor<Integer, Integer> itemProcessor() {
return item -> {
System.out.println("processing item " + item);
return item;
};
}
@Bean
public ItemWriter<Integer> itemWriter() {
return items -> {
for (Integer item : items) {
System.out.println("writing item " + item);
}
};
}
@Bean
public IntegrationFlow workerIntegrationFlow() {
return this.remoteChunkingWorkerBuilder
.itemProcessor(itemProcessor())
.itemWriter(itemWriter())
.inputChannel(requests())
.outputChannel(replies())
.build();
}



3.5、启动Master&Worker



为了查看效果,我先执行了package命令,打了一个可执行jar包,然后分别启动master和woker



启动Master



java -jar fucking-great-springbatch-0.0.1-SNAPSHOT.jar --spring.profiles.active=master --server.port=8080



启动Worker



java -jar fucking-great-springbatch-0.0.1-SNAPSHOT.jar --spring.profiles.active=worker --server.port=8081



3.6、调用接口测试



通过执行如下命令或者通过浏览器打开如下地址



wget http://localhost:8080/launchRemoteChunkingJob



执行完成之后,通过日志我们可以看到master和worker分别有相应日志输出,worker端负责消费



4、附录



https://docs.spring.io/spring-batch/docs/current/reference/html/scalability.html#remoteChunking



https://docs.spring.io/spring-batch/docs/current/reference/html/spring-batch-integration.html#remote-chunking

来玩啊...代码扫上面二维码



发布于: 2020 年 06 月 28 日 阅读数: 42
用户头像

稻草鸟人

关注

悟以往之不谏,知来着之可追 2019.09.24 加入

林花谢了春红,太匆匆。无奈朝来寒雨,晚来风。 胭脂泪,相留醉,几时重。自是人生长恨,水长东。

评论 (2 条评论)

发布
用户头像
有张图片不显示,请更换
2020 年 06 月 28 日 16:58
回复
已更新
2020 年 06 月 28 日 17:35
回复
没有更多了
SpringBatch系列之Remote-Chunking