SpringBatch系列之Remote-Chunking

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

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

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