1、概要
前面的文章介绍了Spring Batch并发并行的批处理能力，但是还不够，单台机器的性能终归有极限，因此我们有些场景就可以考虑使用多台机器来处理。
本文我们将介绍remote chunking，第一篇简单介绍Spring Batch多机器处理披露任务的能力。
2、什么是remote chunking
remote chunking将数据读和写拆分到一个master多个slave机器上。master机器负责读数据并且分发数据到slave机器。master机器在Step中读取数据，并通过像JMS这样的技术将块处理部分交给salve机器。
image
在master端，RemoteChunkingManagerStepBuilderFactory允许我们通过声明如下内容配置master步骤
- 配置
item reader读取数据发送给workers - 配置
output channel(Outgoing requests)发送请求给workers - 配置
input channel(Incoming replies)接收workers响应
没有必要显式声明ChunkMessageChannelItemWriter和MessagingTemplate默认即可（如果需要也可以以显式配置出来）
在worker端，RemoteChunkingWorkerBuilder允许有如下配置
- 通过
input channel(Incoming requests)监听master端发出的请求 - 为每一个请求调用
ChunkProcessorChunkHandler的handleChunk方法执行配置好的ItemProcessor和ItemWriter - 通过
output channel (Outgoing replies)发送响应到master端
没有必要显式声明SimpleChunkProcessor和ChunkProcessorChunkHandler默认即可（如有必要也可以显式配置出来）
从４.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
来玩啊...代码扫下面二维码
网友评论