SpringBatch系列之Remote-Chunking

1、概要

前面的文章介紹了Spring Batch并發(fā)并行的批處理能力，但是還不夠，單臺(tái)機(jī)器的性能終歸有極限，因此我們有些場(chǎng)景就可以考慮使用多臺(tái)機(jī)器來(lái)處理。

本文我們將介紹remote chunking，第一篇簡(jiǎn)單介紹Spring Batch多機(jī)器處理披露任務(wù)的能力。

2、什么是remote chunking

remote chunking將數(shù)據(jù)讀和寫拆分到一個(gè)master多個(gè)slave機(jī)器上。master機(jī)器負(fù)責(zé)讀數(shù)據(jù)并且分發(fā)數(shù)據(jù)到slave機(jī)器。master機(jī)器在Step中讀取數(shù)據(jù)，并通過(guò)像JMS這樣的技術(shù)將塊處理部分交給salve機(jī)器。

image

在master端，RemoteChunkingManagerStepBuilderFactory允許我們通過(guò)聲明如下內(nèi)容配置master步驟

  • 配置item reader讀取數(shù)據(jù)發(fā)送給workers
  • 配置output channel(Outgoing requests)發(fā)送請(qǐng)求給workers
  • 配置input channel(Incoming replies)接收workers響應(yīng)

沒(méi)有必要顯式聲明ChunkMessageChannelItemWriter和MessagingTemplate默認(rèn)即可（如果需要也可以以顯式配置出來(lái)）

在worker端，RemoteChunkingWorkerBuilder允許有如下配置

  • 通過(guò)input channel(Incoming requests)監(jiān)聽(tīng)master端發(fā)出的請(qǐng)求
  • 為每一個(gè)請(qǐng)求調(diào)用ChunkProcessorChunkHandler的handleChunk方法執(zhí)行配置好的ItemProcessor和ItemWriter
  • 通過(guò)output channel (Outgoing replies)發(fā)送響應(yīng)到master端

沒(méi)有必要顯式聲明SimpleChunkProcessor和ChunkProcessorChunkHandler默認(rèn)即可（如有必要也可以顯式配置出來(lái)）

從４.1版本開(kāi)始，Spring Batch Integration通過(guò)注解@EnableBatchIntegration簡(jiǎn)化了remote chunking步驟。這個(gè)注解主要作用是方便注入如下兩個(gè)bean

  • RemoteChunkingManagerStepBuilderFactory: 在master端配置
  • RemoteChunkingWorkerBuilder:用于配置worker端處理流程

3、開(kāi)始多進(jìn)程之旅

3.1、添加依賴

在之前的maven配置基礎(chǔ)之上，添加如下依賴

<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、啟動(dòng)Master＆Worker

為了查看效果，我先執(zhí)行了package命令，打了一個(gè)可執(zhí)行jar包，然后分別啟動(dòng)master和woker

啟動(dòng)Master

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

啟動(dòng)Worker

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

3.6、調(diào)用接口測(cè)試

通過(guò)執(zhí)行如下命令或者通過(guò)瀏覽器打開(kāi)如下地址

wget http://localhost:8080/launchRemoteChunkingJob

執(zhí)行完成之后，通過(guò)日志我們可以看到master和worker分別有相應(yīng)日志輸出，worker端負(fù)責(zé)消費(fèi)

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

來(lái)玩啊...代碼掃下面二維碼
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請(qǐng)聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請(qǐng)結(jié)合常識(shí)與多方信息審慎甄別。
平臺(tái)聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡(jiǎn)書系信息發(fā)布平臺(tái)，僅提供信息存儲(chǔ)服務(wù)。

友情鏈接更多精彩內(nèi)容