1. 背景
上章讲到了并行处理相关内容。但远程分区还是讲的不够细。还差
远程分区的db轮询模式
分区器Partitioner 和PartitionHandler的使用，
聚合器的使用。
远程分区的整体流程
2. 远程分区的db轮询模式
- Master用户端需要实现 outgoIng,outband
@Bean //定义channle
public DirectChannel managerDBPollRequests() {
return new DirectChannel();
}
@Bean // 定义从channel内容经过amqp发送到requests队列中
public IntegrationFlow managerDBPollOutboundFlow() {
return IntegrationFlows.from(managerDBPollRequests())
.handle(Amqp.outboundAdapter(rabbitTemplate).routingKey("requests"))
.get();
}
// 定义远程分区step Manager端(master)
@Bean
public Step managerDBPollStep() {
return this.managerStepBuilderFactory.get("managerDBPollStep") // 定义名称
.partitioner("workerDBPollStep", new SimplePartitioner()) //传入分区器
.gridSize(GRID_SIZE) // 传入分区器可以分几个区
.outputChannel(managerDBPollRequests()) //将分区结果发送到对应的channel
.build();
}
代码流程大概是上图的样子。
-
worker用户端需要实现 incomingRequest,inband
// 定义消息接收channel @Bean public DirectChannel workerDBPollRequests() { return new DirectChannel(); } @Bean public IntegrationFlow workerDBPollInboundFlow(ConnectionFactory rabbitmqConnectionFactory) { return IntegrationFlows .from(Amqp.inboundAdapter( rabbitmqConnectionFactory,"requests"))//从消息队列request来消息 .channel(workerDBPollRequests()//执行完的消息内容，发送到channel ).get(); } @Bean public Step workerDBPollStep() { return this.workerStepBuilderFactory .get("workerDBPollStep") //step名称 .inputChannel(workerDBPollRequests()) // 接收channel .tasklet(workerDBPollStepTasklet(null)) .build(); }
3. 分区器Partitioner 和PartitionHandler的使用
- PartitionHandler 分区执行器
这PartitionHandler是了解远程处理或网格环境结构的组件。它能够向StepExecution远程实例发送请求Step ，以某种特定于结构的格式（如 DTO）进行包装。它不必知道如何拆分输入数据或如何聚合多次Step执行的结果。一般来说，它可能也不需要了解弹性或故障转移，因为在许多情况下，这些都是结构的特性。在任何情况下，Spring Batch 始终提供独立于结构的可重启性。失败的Job总是可以重新启动，只有失败Steps的才会重新执行。
该PartitionHandler接口可以为各种结构类型提供专门的实现，包括简单的 RMI 远程处理、EJB 远程处理、自定义 Web 服务、JMS、Java 空间、共享内存网格（如 Terracotta 或 Coherence）和网格执行结构（如 GridGain）。Spring Batch 不包含任何专有网格或远程结构的实现。
然而，Spring Batch 确实提供了一个有用的实现，它使用Spring 的策略在单独的执行线程中本地PartitionHandler执行Step实例 。TaskExecutor该实现称为 TaskExecutorPartitionHandler.
可以在TaskExecutorPartitionHandlerjava 配置中显式配置，如下例所示：
@Bean
public Step step1Manager() {
return stepBuilderFactory.get("step1.manager")
.partitioner("step1", new SimplePartitioner())
.partitionHandler(partitionHandler())
.build();
}
@Bean
public PartitionHandler partitionHandler() {
TaskExecutorPartitionHandler retVal = new TaskExecutorPartitionHandler();
retVal.setTaskExecutor(new SimpleAsyncTaskExecutor());
retVal.setStep(step1());
retVal.setGridSize(10);
return retVal;
}
gridSize属性确定要创建的单独步骤执行的数量，因此它可以与TaskExecutor中线程池的大小相匹配。或者，可以将其设置为大于可用的线程数，这使得工作块更小。
TaskExecutionPartitionHandler对于IO密集型步骤实例非常有用，例如复制大量文件或将文件系统复制到内容管理系统中。它还可以通过提供作为远程调用代理的步骤实现（例如使用Spring远程处理）来用于远程执行。
- 分区器Partitioner
有一个更简单的Partitioner职责：仅生成执行上下文作为新步骤执行的输入参数（无需担心重新启动）。它只有一个方法，如下面的接口定义所示：
public interface Partitioner {
Map<String, ExecutionContext> partition(int gridSize);
}
此方法的返回值将每个步骤执行的唯一名称 (Map中的key) ，值为具体的step的ExecutionContext信息。这些名称稍后在批处理元数据中显示为分区中的步骤名称StepExecutions。它 ExecutionContext只是一个名称-值对的包，因此它可能包含一系列主键、行号或输入文件的位置。然后远程Step通常使用占位符绑定到上下文输入#{…} 使用stepScope 中的@Value来获取
{partition2={start=20}, partition1={start=10}, partition0={start=1}}
4. StepExecutionAggregator聚合器的使用
-
StepExecutionAggregator 最后的统计分析聚合逻辑
void aggregate(StepExecution result, Collection<StepExecution> executions);该接口只有一个方法。一个方法从入参来看是。接受Collection<StepExecution> executions 的集合。也就是和partitioner的反过程。分而治之的思想，汇总聚合分区step完成manager状态的更新。
默认实现如下 DefaultStepExecutionAggregator.java
-
public void aggregate(StepExecution result, Collection<StepExecution> executions) { Assert.notNull(result, "To aggregate into a result it must be non-null."); if (executions == null) { return; } for (StepExecution stepExecution : executions) { BatchStatus status = stepExecution.getStatus(); result.setStatus(BatchStatus.max(result.getStatus(), status)); result.setExitStatus(result.getExitStatus().and(stepExecution.getExitStatus())); result.setFilterCount(result.getFilterCount() + stepExecution.getFilterCount()); result.setProcessSkipCount(result.getProcessSkipCount() + stepExecution.getProcessSkipCount()); result.setCommitCount(result.getCommitCount() + stepExecution.getCommitCount()); result.setRollbackCount(result.getRollbackCount() + stepExecution.getRollbackCount()); result.setReadCount(result.getReadCount() + stepExecution.getReadCount()); result.setReadSkipCount(result.getReadSkipCount() + stepExecution.getReadSkipCount()); result.setWriteCount(result.getWriteCount() + stepExecution.getWriteCount()); result.setWriteSkipCount(result.getWriteSkipCount() + stepExecution.getWriteSkipCount()); } }这个aggregate是在什么时候设置进去的呢 PartitionStep#doExecute
protected void doExecute(StepExecution stepExecution) throws Exception { stepExecution.getExecutionContext().put(STEP_TYPE_KEY, this.getClass().getName()); // Wait for task completion and then aggregate the results Collection<StepExecution> executions = partitionHandler.handle(stepExecutionSplitter, stepExecution); stepExecution.upgradeStatus(BatchStatus.COMPLETED); stepExecutionAggregator.aggregate(stepExecution, executions); // If anything failed or had a problem we need to crap out if (stepExecution.getStatus().isUnsuccessful()) { throw new JobExecutionException("Partition handler returned an unsuccessful step"); } }分区step如何设置aggregate
@Bean
public Step step1Manager() {
return stepBuilderFactory.get("step1.manager")
.partitioner("step1", new SimplePartitioner())
.partitionHandler(partitionHandler())
.aggregator(new DefaultStepExecutionAggregator())
.build();
}
5. 整体流程
整体流程就是partitionStep的执行过程
protected void doExecute(StepExecution stepExecution) throws Exception {
stepExecution.getExecutionContext().put(STEP_TYPE_KEY, this.getClass().getName());
// Wait for task completion and then aggregate the results
// 等待任务结束然后聚合结果。
Collection<StepExecution> executions = partitionHandler.handle(stepExecutionSplitter, stepExecution);
// 不管结果是成功还是失败，直接更新结果为完成
stepExecution.upgradeStatus(BatchStatus.COMPLETED);
// 调用聚合器，将分区结果聚合
stepExecutionAggregator.aggregate(stepExecution, executions);
// If anything failed or had a problem we need to crap out
// 聚合的结果如果不是成功状态，那么直接抛出错误。step异常结束，job异常结束
if (stepExecution.getStatus().isUnsuccessful()) {
throw new JobExecutionException("Partition handler returned an unsuccessful step");
}
}
// 状态为failed 或者 大于FAILED,即 ABANDONED, UNKNOWN都为失败
public boolean isUnsuccessful() {
return this == FAILED || this.isGreaterThan(FAILED);
}
