SpringBatch 系列之 Remote-Chunking

关注
发布于: 2020 年 06 月 28 日
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响应
﻿
没有必要显式声明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
﻿
https://docs.spring.io/spring-batch/docs/current/reference/html/spring-batch-integration.html#remote-chunking
来玩啊...代码扫上面二维码
﻿
发布于: 2020 年 06 月 28 日阅读数: 80
原文链接:【http://xie.infoq.cn/article/6dec472e85ad0ae6dd5055ba9】。
稻草鸟人

关注
悟以往之不谏，知来着之可追 2019.09.24 加入
林花谢了春红，太匆匆。无奈朝来寒雨，晚来风。胭脂泪，相留醉，几时重。自是人生长恨，水长东。
评论 (2 条评论)

发布
InfoQ写作平台
有张图片不显示，请更换
2020 年 06 月 28 日 16:58
 0 回复
稻草鸟人
已更新
2020 年 06 月 28 日 17:35
 0 回复
没有更多了
创作场景

SpringBatch 系列之 Remote-Chunking

1、概要

2、什么是`remote chunking`

3、开始多进程之旅

3.1、添加依赖

3.2、配置ActiveMQ

3.3、Master端程序

3.4、Worker端程序

3.5、启动Master＆Worker

3.6、调用接口测试

4、附录

稻草鸟人

评论 (2 条评论)
