Spring批处理远程分块

  Spring Batch的远程分块其实是一种主从分布式处理模式,一个主处理机和一个或多个从处理机,这样提高了批处理的计算能力,主从之间是通过消息中间件JMS进行通信。

在远程分块中,Step处理分为主从多个处理过程,主从之间通过一些中间件相互通信。下图显示了该模式:

master主组件是一个单个处理,slave从属组件则是多个远程处理过程。如果主处理不会造成瓶颈,这种模式效果最好,因为一般情况下输出处理必须比读取处理更昂贵(在实践中通常是这种情况)。

主master组件知道如何使用消息的通用版本发送条目item块到消息中间件,slave从属服务器是消息中间件的标准监听器(例如如果使用JMS,则是MesssageListener实现),然后调用ItemWriter或ItemProcessor 来处理条目块。

使用此模式的一个优点是读取器、处理器和输出器等组件是现成的(与用于本地执行步骤的组件相同)。这些项目是动态划分的,工作通过中间件共享,因此,如果监听器都是繁忙消费者,那么将自动加入负载平衡。

Spring Batch集成

   Spring批处理通常是和消息系统等中间件集成的,向批处理流程添加消息传递可实现操作的自动化,也可实现方式和策略的分离,比如消息触发要执行的作业,然后以各种方式发送消息。或者,当作业完成或失败时,会触发发送的相应成功或失败的消息等等,消息的使用者可能与批处理应用本身无关。

  消息传递也可以嵌入到批处理的作业执行中,远程分区和远程分块提供了将工作负载分配给多个处理器工作。

本项目我们使用spring-batch-integration实现远程分块。

这里我们使用ActiveMQ作为JMS的实现,主处理和从处理之间通过两个队列通讯,requests队列时主节点通过chunkMessageChannelItemWriter发往从节点的,而从节点通过ChunkProcessorChunkHandler接受到消息后,再将结果发往主节点的消息通过replies。

主节点的Job代码如下:


@Autowired
private JobBuilderFactory jobBuilderFactory;
@Autowired
private StepBuilderFactory stepBuilderFactory;
@Bean
public TaskletStep masterStep() {
return this.stepBuilderFactory.get("masterStep")
.<Integer, Integer>chunk(3)
.reader(itemReader())
.writer(itemWriter())
.build();
}

@Bean
public Job remoteChunkingJob() {
return this.jobBuilderFactory.get("remoteChunkingJob")
.start(masterStep())
.build();
} @Bean
public ItemReader<Integer> itemReader() {
return new ListItemReader<>(Arrays.asList(1, 2, 3, 4, 5, 6));
}

@Bean
public ItemWriter<Integer> itemWriter() {
MessagingTemplate messagingTemplate = new MessagingTemplate();
messagingTemplate.setDefaultChannel(requests());
ChunkMessageChannelItemWriter<Integer> chunkMessageChannelItemWriter = new ChunkMessageChannelItemWriter<>();
chunkMessageChannelItemWriter.setMessagingOperations(messagingTemplate);
chunkMessageChannelItemWriter.setReplyChannel(replies());
return chunkMessageChannelItemWriter;
}

与JMS连接的代码如下:

@Value("${broker.url}")
private String brokerUrl;


@Bean
public ActiveMQConnectionFactory jmsConnectionFactory() {
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 jmsConnectionFactory) {
return IntegrationFlows
.from(requests())
.handle(Jms.outboundAdapter(jmsConnectionFactory).destination("requests"))
.get();
}

/*
* Configure inbound flow (replies coming from workers)
*/

@Bean
public QueueChannel replies() {
return new QueueChannel();
}

@Bean
public IntegrationFlow inboundFlow(ActiveMQConnectionFactory jmsConnectionFactory) {
return IntegrationFlows
.from(Jms.messageDrivenChannelAdapter(jmsConnectionFactory).destination("replies"))
.channel(replies())
.get();
}

Application.properties:

broker.url=tcp://localhost:61616

spring.batch.initialize-schema=always

spring.datasource.url=jdbc:mysql://localhost:3306/mytest
spring.datasource.username=root
spring.datasource.password=root
spring.datasource.driver-class-name=com.mysql.jdbc.Driver

从节点的Job如下:

/*
* 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 + System.currentTimeMillis());
}
};
}

@Bean
@ServiceActivator(inputChannel = "requests", outputChannel = "replies")
public ChunkProcessorChunkHandler<Integer> chunkProcessorChunkHandler() {
ChunkProcessor<Integer> chunkProcessor = new SimpleChunkProcessor<>(itemProcessor(), itemWriter());
ChunkProcessorChunkHandler<Integer> chunkProcessorChunkHandler = new ChunkProcessorChunkHandler<>();
chunkProcessorChunkHandler.setChunkProcessor(chunkProcessor);
return chunkProcessorChunkHandler;
}

从节点连接JMS代码类似于主节点,自动实现了JMS的侦听器作用。从节点主要是在chunkProcessorChunkHandler这里做了处理和输出回到主节点。

本节源码:Github

Spring batch

 

 

猜你喜欢