SpringBatch系列之Remote

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配置基础之上，添加如下依赖

1<dependency>
2            <groupId>org.springframework.boot</groupId>
3            <artifactId>spring-boot-starter-amqp</artifactId>
4        </dependency>
5        <dependency>
6            <groupId>org.springframework.boot</groupId>
7            <artifactId>spring-boot-starter-integration</artifactId>
8        </dependency>
9        <dependency>
10            <groupId>org.springframework.batch</groupId>
11            <artifactId>spring-batch-integration</artifactId>
12        </dependency>
13        <dependency>
14            <groupId>org.springframework.boot</groupId>
15            <artifactId>spring-boot-starter-activemq</artifactId>
16        </dependency>
17        <dependency>
18            <groupId>org.springframework.integration</groupId>
19            <artifactId>spring-integration-jms</artifactId>
20        </dependency>

3.2、配置ActiveMQ

broker.url=tcp://localhost:61616

3.3、Master端程序

1@Profile("master")
2public class ManagerConfiguration {
3
4    @Value("${broker.url}")
5    private String brokerUrl;
6
7    @Autowired
8    private JobBuilderFactory jobBuilderFactory;
9
10    @Autowired
11    private RemoteChunkingManagerStepBuilderFactory managerStepBuilderFactory;
12
13    @Bean
14    public ActiveMQConnectionFactory connectionFactory() {
15        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
16        connectionFactory.setBrokerURL(this.brokerUrl);
17        connectionFactory.setTrustAllPackages(true);
18        return connectionFactory;
19    }
20
21    /*
22     * Configure outbound flow (requests going to workers)
23     */
24    @Bean
25    public DirectChannel requests() {
26        return new DirectChannel();
27    }
28
29    @Bean
30    public IntegrationFlow outboundFlow(ActiveMQConnectionFactory connectionFactory) {
31        return IntegrationFlows
32                .from(requests())
33                .handle(Jms.outboundAdapter(connectionFactory).destination("requests"))
34                .get();
35    }
36
37    /*
38     * Configure inbound flow (replies coming from workers)
39     */
40    @Bean
41    public QueueChannel replies() {
42        return new QueueChannel();
43    }
44
45    @Bean
46    public IntegrationFlow inboundFlow(ActiveMQConnectionFactory connectionFactory) {
47        return IntegrationFlows
48                .from(Jms.messageDrivenChannelAdapter(connectionFactory).destination("replies"))
49                .channel(replies())
50                .get();
51    }
52
53    /*
54     * Configure master step components
55     */
56    @Bean
57    public ListItemReader<Integer> itemReader() {
58        return new ListItemReader<>(Arrays.asList(1, 2, 3, 4, 5, 6));
59    }
60
61    @Bean
62    public TaskletStep managerStep() {
63        return this.managerStepBuilderFactory.get("managerStep")
64                .<Integer, Integer>chunk(3)
65                .reader(itemReader())
66                .outputChannel(requests())
67                .inputChannel(replies())
68                .build();
69    }
70
71    @Bean("remoteChunkingJob")
72    public Job remoteChunkingJob() {
73        return this.jobBuilderFactory.get("remoteChunkingJob")
74                .start(managerStep())
75                .build();
76    }
77}

3.4、Worker端程序

1@Profile("worker")
2public class WorkerConfiguration {
3    @Value("${broker.url}")
4    private String brokerUrl;
5
6    @Resource
7    private RemoteChunkingWorkerBuilder<Integer, Integer> remoteChunkingWorkerBuilder;
8
9    @Bean
10    public ActiveMQConnectionFactory connectionFactory() {
11        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory();
12        connectionFactory.setBrokerURL(this.brokerUrl);
13        connectionFactory.setTrustAllPackages(true);
14        return connectionFactory;
15    }
16
17    /*
18     * Configure inbound flow (requests coming from the master)
19     */
20    @Bean
21    public DirectChannel requests() {
22        return new DirectChannel();
23    }
24
25    @Bean
26    public IntegrationFlow inboundFlow(ActiveMQConnectionFactory connectionFactory) {
27        return IntegrationFlows
28                .from(Jms.messageDrivenChannelAdapter(connectionFactory).destination("requests"))
29                .channel(requests())
30                .get();
31    }
32
33    /*
34     * Configure outbound flow (replies going to the master)
35     */
36    @Bean
37    public DirectChannel replies() {
38        return new DirectChannel();
39    }
40
41    @Bean
42    public IntegrationFlow outboundFlow(ActiveMQConnectionFactory connectionFactory) {
43        return IntegrationFlows
44                .from(replies())
45                .handle(Jms.outboundAdapter(connectionFactory).destination("replies"))
46                .get();
47    }
48
49    /*
50     * Configure worker components
51     */
52    @Bean
53    public ItemProcessor<Integer, Integer> itemProcessor() {
54        return item -> {
55            System.out.println("processing item " + item);
56            return item;
57        };
58    }
59
60    @Bean
61    public ItemWriter<Integer> itemWriter() {
62        return items -> {
63            for (Integer item : items) {
64                System.out.println("writing item " + item);
65            }
66        };
67    }
68
69    @Bean
70    public IntegrationFlow workerIntegrationFlow() {
71        return this.remoteChunkingWorkerBuilder
72                .itemProcessor(itemProcessor())
73                .itemWriter(itemWriter())
74                .inputChannel(requests())
75                .outputChannel(replies())
76                .build();
77    }

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