Schedulerx2.0工做流支持数据传输

时间 2019-11-24

标签 schedulerx2.0 schedulerx 支持数据传输栏目系统网络繁體版

原文原文链接

1. 前言

Schedulerx2.0是阿里中间件自研的基于akka架构的新一代分布式任务调度平台，提供定时、任务编排、分布式跑批等功能，具备高可靠、海量任务、秒级调度等能力。java

Schedulerx2.0提供可视化的工做流进行任务编排，该文章将详细介绍如何使用schedulerx2.0的工做流进行上下游任务的数据传输。架构

2. 接口介绍

2.1 支持的执行方式和任务类型

当前只有java任务支持数据传输，网格计算请使用MapReduce模型进行数据传输。less

2.2 返回执行结果

/**
 *
 * @param status
 * @param result, the size should less than 1000 bytes
 * @throws Exception
 */
public ProcessResult(boolean status, String result) throws Exception;

在Processor结尾，经过该结果替代ProcessResult(boolean status)，能够返回执行结果。分布式

result的长度不能超过1000个字节（注意，不是String的长度，若是有中文字符，可能会超过1000个字节！），若是超过1000个字节，任务会失败。ide

2.3 获取上游数据

List<JobInstanceData> upstreamDatas = JobContext.getUpstreamData();

在Processor里，能够经过该接口从JobContext中拿到上游的数据。上游的数据是一个list(可能有多个父节点)，JobInstanceData里有两个属性，分别是jobName和data(String类型)。url

3. Demo演示

首先咱们写三个jobProcessorspa

public class TestSimpleJobA extends JavaProcessor {
    @Override
    public ProcessResult process(JobContext context) throws Exception {
        System.out.println("TestSimpleJobA " + DateTime.now().toString("yyyy-MM-dd HH:mm:ss"));
        return new ProcessResult(true, String.valueOf(1));
    }
}

public class TestSimpleJobB extends JavaProcessor {
    @Override
    public ProcessResult process(JobContext context) throws Exception {
        System.out.println("TestSimpleJobB " + DateTime.now().toString("yyyy-MM-dd HH:mm:ss"));
        return new ProcessResult(true, String.valueOf(2));
    }
}

public class TestSimpleJobC extends JavaProcessor {

    @Override
    public ProcessResult process(JobContext context) throws Exception {
        List<JobInstanceData> upstreamDatas = context.getUpstreamData();
        int sum = 0;
        for (JobInstanceData jobInstanceData : upstreamDatas) {
            System.out.println("jobName=" + jobInstanceData.getJobName() + ", data=" + jobInstanceData.getData());
            sum += Integer.valueOf(jobInstanceData.getData());
        }
        System.out.println("TestSimpleJobC sum=" + sum);
        return new ProcessResult(true, String.valueOf(sum));
    }

}

经过控制台配置工做流以下图所示：code

触发一次该工做流，而后进入工做流实例图，右键jobA的实例，进入详情，能够看到jobA实例结果=1，以下图中间件

同理，能够看到jobB的实例结果=2， jobC的实例结果=3blog

控制台也能看到jobC的机器打印

jobName=jobB, data=2
jobName=jobA, data=1
TestSimpleJobC sum=3

原文连接本文为云栖社区原创内容，未经容许不得转载。