Schedulerx2.0分布式执行之——广播执行

2021-11-08 07:00:56

1. 简介

广播执行表示一个任务实例会广播到该分组所有worker上执行，当所有机器都执行完成，该实例才算完成。

任意一台worker执行失败，都算该实例失败。

所有worker执行成功，才算该实例成功。

有子任务列表，可以看每台机器的执行详情。

2. 执行方式

执行方式选择广播

3. 任务类型

任务类型可以选择多种，比如脚本，或者java任务。如果选择java，还支持preProcess和postProcess高级特性。

使用java任务需要继承JavaProcessor（1.0.8+版本），接口如下：

public ProcessResult process(JobContext context) throws Exception; (必选)
public void preProcess(JobContext context); （可选）
public ProcessResult postProcess(JobContext context); （可选）

preProcess会在所有机器执行process之前执行，只会执行一次。
postProcess会在所有机器执行process之后且都成功执行，之后执行一次，可以返回结果，作为工作流数据传输。

4. 应用场景

4.1 批量运维

定时广播所有机器运行某个脚本。
定时广播所有机器清理缓存。

4.2 数据聚合

1）使用BroadcastJavaProcessor，preProcess的时候重置数据库count=0；
2）每台机器执行process的时候，根据自己机器内存或者日志的值，往数据库count叠加；
3）postProcess的时候，读取数据库count值，传给下游；

5. Demo

@Component
public class TestBroadcastJobProcessor extends JavaProcessor {

    @Override
    public ProcessResult process(JobContext context) throws Exception {
        System.out.println("this is process");
        return new ProcessResult(true);
    }

    @Override
    public void preProcess(JobContext context) {
        System.out.println("this is preProcess");
    }

    @Override
    public ProcessResult postProcess(JobContext context) {
        System.out.println("this is postProcess");
        return new ProcessResult(true, "hello broadcast");
    }
}

如上代码所示，起三台机器，执行结果如下：
worker1:

this is preProcess
this is process
this is postProcess

worker2:

this is process

worker3:

this is process

码农公寓