MapReduce 模型是 Map 模型的扩展,新增 Reduce接口,需要实现 MapReduceJobProcessor。

注意事项

  • MapReduce 模型只有一个 Reduce,所有子任务完成后会执行 Reduce 方法,可以在 Reduce 方法中返回该任务示例的执行结果,作为工作流的上下游数据传递。如果有子任务失败,Reduce 不会执行。Reduce 失败,整个任务示例也失败。
  • Scheduler X 不保证子任务一定执行一次,在特殊条件下会 failover,可能会导致子任务重复执行,需要业务方自己实现幂等。
  • Scheduler X 使用的是 Hessian 序列化框架,目前不支持 LocalDateTimeBigDecimal。子任务中如果有如上两个数据结构,请替换其他的数据结构(特别是 BigDecimal,序列化不会报错,反序列化会变成 0)。

接口

接口 解释 是否必选
ProcessResult process(JobContext context) 每个子任务执行业务的入口,需要从 context 里获取 taskName,自己判断是哪个子任务,进行相应的逻辑处理。执行完成后,需要返回 ProcessResult。ProcessResult 接口请参见 ProcessResult
ProcessResult map(List<? extends Object> taskList, String taskName) 执行 map 方法可以把一批子任务分布式到多台机器上执行,可以 map 多次。如果 taskList 是空,返回失败。执行完成后,需要返回 ProcessResult。ProcessResult 接口请参见 ProcessResult
void kill(JobContext context) 前端 kill 任务会触发该方法,需要用户自己实现如何中断业务。

执行方式

  • 并行计算:最多支持 300 任务,有子任务列表。
    注意 秒级别任务不要选择并行计算。
  • 内存网格:基于内存计算,最多支持 50,000 以下子任务,速度快。
  • 网格计算:基于文件计算,最多支持 1,000,000 子任务。

高级配置

高级配置请参见任务管理高级配置参数说明

发送 500 条消息的 Demo 示例(适用于 MapReduce 模型)

@Component
public class TestMapReduceJobProcessor extends MapReduceJobProcessor {

    @Override
    public ProcessResult process(JobContext context) throws Exception {
        String taskName = context.getTaskName();
        int dispatchNum=500;
        if (isRootTask(context)) {
            System.out.println("start root task");
            List<String> msgList = Lists.newArrayList();
            for (int i = 0; i <= dispatchNum; i++) {
                msgList.add("msg_" + i);
            }
            return map(msgList, "Level1Dispatch");
        } else if (taskName.equals("Level1Dispatch")) {
            String task = (String)context.getTask();
            System.out.println(task);
            return new ProcessResult(true);
        }

        return new ProcessResult(false);
    }

    @Override
    public ProcessResult reduce(JobContext context) throws Exception {
        return new ProcessResult(true, "TestMapReduceJobProcessor.reduce");
    }

}           

处理单表数据的 Demo 示例(适用于 Map 或 MapReduce 模型)

@Component
public class ScanSingleTableJobProcessor extends MapJobProcessor {

    @Service
    private XXXService xxxService;

    private final int PAGE_SIZE = 500;

    static class PageTask {
        private long startId;
        private long endId;

        public PageTask(long startId, long endId) {
            this.startId = startId;
            this.endId = endId;
        }

        public long getStartId() {
            return startId;
        }

        public long getEndId() {
            return endId;
        }
    }

    @Override
    public ProcessResult process(JobContext context) throws Exception {
        String tableName = context.getJobParameters(); //多个 Job 后端代码可以一致,通过控制台配置 Job 参数表示表名。
        String taskName = context.getTaskName();
        Object task = context.getTask();
        if (isRootTask(context)) {
            Pair<Long, Long> idPair = queryMinAndMaxId(tableName);
            long minId = idPair.getFirst();
            long maxId = idPair.getSecond();
            List<PageTask> tasks = Lists.newArrayList();
            int step = (int) ((maxId - minId) / PAGE_SIZE); //计算分页数量
            for (long i = minId; i < maxId; i+=step) {
                tasks.add(new PageTask(i, (i+step > maxId ? maxId : i+step)));
            }
            return map(tasks, "PageTask");
        } else if (taskName.equals("PageTask")) {
            PageTask pageTask = (PageTask)task;
            long startId = pageTask.getStartId();
            long endId = pageTask.getEndId();
            List<Record> records = queryRecord(tableName, startId, endId);
            //TODO handle records
            return new ProcessResult(true);
        }

        return new ProcessResult(false);
    }

    private Pair<Long, Long> queryMinAndMaxId(String tableName) {
        //TODO select min(id),max(id) from [tableName]
        return new Pair<Long, Long>(1L, 10000L);
    }

    private List<Record> queryRecord(String tableName, long startId, long endId) {
        List<Record> records = Lists.newArrayList();
        //TODO select * from [tableName] where id>=[startId] and id<[endId]
        return records;
    }

}            

处理分库分表数据的 Demo 示例(适用于 Map 或 MapReduce 模型)

@Component
public class ScanShardingTableJobProcessor extends MapJobProcessor {

    @Service
    private XXXService xxxService;

    private final int PAGE_SIZE = 500;

    static class PageTask {
        private String tableName;
        private long startId;
        private long endId;

        public PageTask(String tableName, long startId, long endId) {
            this.tableName = tableName;
            this.startId = startId;
            this.endId = endId;
        }

        public String getTableName() {
            return tableName;
        }

        public long getStartId() {
            return startId;
        }

        public long getEndId() {
            return endId;
        }
    }

    @Override
    public ProcessResult process(JobContext context) throws Exception {
        String taskName = context.getTaskName();
        Object task = context.getTask();
        if (isRootTask(context)) {
            //先分库
            List<String> dbList = getDbList();
            return map(dbList, "DbTask");
        } else if (taskName.equals("DbTask")) {
            //根据分库去分表
            String dbName = (String)task;
            List<String> tableList = getTableList(dbName);
            return map(tableList, "TableTask");
        } else if (taskName.equals("TableTask")) {
            //如果一个分表也很大,再分页
            String tableName = (String)task;
            Pair<Long, Long> idPair = queryMinAndMaxId(tableName);
            long minId = idPair.getFirst();
            long maxId = idPair.getSecond();
            List<PageTask> tasks = Lists.newArrayList();
            int step = (int) ((maxId - minId) / PAGE_SIZE); //计算分页数量
            for (long i = minId; i < maxId; i+=step) {
                tasks.add(new PageTask(tableName, i, (i+step > maxId ? maxId : i+step)));
            }
            return map(tasks, "PageTask");
        } else if (taskName.equals("PageTask")) {
            PageTask pageTask = (PageTask)task;
            String tableName = pageTask.getTableName();
            long startId = pageTask.getStartId();
            long endId = pageTask.getEndId();
            List<Record> records = queryRecord(tableName, startId, endId);
            //TODO handle records
            return new ProcessResult(true);
        }

        return new ProcessResult(false);
    }

    private List<String> getDbList() {
        List<String> dbList = Lists.newArrayList();
        //TODO 返回分库列表
        return dbList;
    }

    private List<String> getTableList(String dbName) {
        List<String> tableList = Lists.newArrayList();
        //TODO 返回分表列表
        return tableList;
    }

    private Pair<Long, Long> queryMinAndMaxId(String tableName) {
        //TODO select min(id),max(id) from [tableName]
        return new Pair<Long, Long>(1L, 10000L);
    }

    private List<Record> queryRecord(String tableName, long startId, long endId) {
        List<Record> records = Lists.newArrayList();
        //TODO select * from [tableName] where id>=[startId] and id<[endId]
        return records;
    }

}            

处理 50 条消息并且返回子任务结果由 Reduce 汇总的 Demo 示例(适用于 MapReduce 模型)

@Component
public class TestMapReduceJobProcessor extends MapReduceJobProcessor {

    @Override
    public ProcessResult process(JobContext context) throws Exception {
        String taskName = context.getTaskName();
        int dispatchNum = 50;
        if (context.getJobParameters() != null) {
            dispatchNum = Integer.valueOf(context.getJobParameters());
        }
        if (isRootTask(context)) {
            System.out.println("start root task");
            List<String> msgList = Lists.newArrayList();
            for (int i = 0; i <= dispatchNum; i++) {
                msgList.add("msg_" + i);
            }
            return map(msgList, "Level1Dispatch");
        } else if (taskName.equals("Level1Dispatch")) {
            String task = (String)context.getTask();
            Thread.sleep(2000);
            return new ProcessResult(true, task);
        }

        return new ProcessResult(false);
    }

    @Override
    public ProcessResult reduce(JobContext context) throws Exception {
        for (Entry<Long, String> result : context.getTaskResults().entrySet()) {
            System.out.println("taskId:" + result.getKey() + ", result:" + result.getValue());
        }
        return new ProcessResult(true, "TestMapReduceJobProcessor.reduce");
    }
}