1. 背景

上章讲到了并行处理相关内容。但远程分区还是讲的不够细。还差

远程分区的db轮询模式

分区器Partitioner 和PartitionHandler的使用,

聚合器的使用。

远程分区的整体流程

2. 远程分区的db轮询模式

Remote Partitioning Configuration (with job repository polling)

  • 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();
    }
    

img

代码流程大概是上图的样子。

  • 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();
        }
    

    img

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 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);
	}
作者:|jackssybin|,原文链接: http://www.imooc.com/article/325948

文章推荐

负载均衡之keepalived

Python常用标准库(pickle序列化和JSON序列化)

『忘了再学』Shell基础 — 19、使用declare命令声明变量类型

为什么强烈不推荐使用stop、suspend方法来中断线程?

python如何实现网络测试,了解一下speedtest-cli...

一二阶段部分整理初稿

Canvas之鼠标滑动特效

就是 Spring 设计了一个lx.gongxuanwang.com

java多线程相关

手把手教你玩多数据源动态切换!

java并发编程-StampedLock高性能读写锁

Java 处理 Exception 的 9 个最佳实践,你做对了吗?