@Overridepublic ProcessResult process(final JobContext jobContext) throws Exception {// 处理master任务if (isRootTask(jobContext)) {return processRootTask(jobContext);}
// 处理分发子任务if (StringUtils.equals(jobContext.getTaskName(), SUB_TASK_NAME)) {return processDressOfflineDataWritingIcTask(jobContext);}
return new ProcessResult(true);}// 执行结果汇总@Overridepublic ProcessResult reduce(final JobContext jobContext) throws Exception {}

/** * 组装离线任务上下文 * * @param jobContext 任务基本信息 * @param context离线任务上下文 */public void assembleContextParam(final JobContext jobContext, final DressWritingIcTaskContext context) {final JSONObject params;try {params = JSON.parseObject(jobContext.getInstanceParameters());} catch (Exception e) {throw new RuntimeException(e);}context.setUpdateType(TaskUpdateTypeEnum.parse(params.getString(UPDATE_TYPE)));context.setTableName(params.getString(ODPS_TABLE));context.setProjectName(params.getString(ODPS_PROJECT));context.setPartition(params.getString(PARTITION));context.setTaskId(jobContext.getTaskId());context.setJobInstanceId(jobContext.getJobInstanceId());}

@Overridepublic ProcessResult process(final JobContext jobContext) throws Exception {// 处理master任务if (isRootTask(jobContext)) {return processRootTask(jobContext);}
// 处理分发子任务if (StringUtils.equals(jobContext.getTaskName(), SUB_TASK_NAME)) {return processDressOfflineDataWritingIcTask(jobContext);}
return new ProcessResult(true);}

/** * 子任务处理主流程 * * @param jobContext 子任务基本信息 * @return 子任务处理结果 */private ProcessResult processdressWritingIcTask(final JobContext jobContext) {// 1.获取子任务上下文final DressWritingIcTaskContext dataWritingIcTask = (DressWritingIcTaskContext)jobContext.getTask();
// 2.处理子任务final TaskUpdateResult taskUpdateResult = processRecordsByPage(dataWritingIcTask);
// 3.返回任务处理结果return new ProcessResult(true, JSONObject.toJSONString(taskUpdateResult));}

/** * 结果汇总发送钉钉以及群机器人通知主流程 * * @param jobContext 任务相关数据 */@Overridepublic ProcessResult reduce(final JobContext jobContext) {final TaskUpdateResult processResult = dressWritingIcProcessManager.getSuccessCountFromProcessResult(jobContext);// 数据组装...// 发送钉钉通知}
/** * 子任务处理结果统计 * * @param jobContext reduce中获取任务基本信息及结果容器 * @return 汇总更新结果 */public TaskUpdateResult getSuccessCountFromProcessResult(JobContext jobContext) {TaskUpdateResult taskUpdateResult = new TaskUpdateResult();for (String value : jobContext.getTaskResults().values()) {if (StringUtils.isNotBlank(value)) {try {// 对子任务执行结果中需要采集的数据进行整合// ...} catch (Exception e) {LoggerUtil.error(logger, e, "Parse taskUpdateResult failed,value:", value);}}}
return taskUpdateResult;}

此前得知，单线程时请求更新耗平均约为40ms，那么单机单线程qps约为25qps，ic商品中心规定qps千级别以下请求不需要考虑限流问题，因为我们规定单机限流100，线上机器共10台，虽然理论值1000qps，实则可能小于1000qps；进行子任务分发时，单机限流100，则规定单机4线程并行处理数据，以一分钟单机处理最大数据量为界限：60 * 1000ms /30 * 4= 8000 条数据，则switch配置的分页阈值为8000；上述提到的40ms是大部分请求下的执行耗时，30ms是较快请求下的执行耗时。

如果odps数据总量totalCount <= 在线机器采用默认分页可以处理的数据量（默认单机分页*在线机器数） 则单机任务处理量为默认界限值, 任务数量由总数量/分页获取；否则（超过界限值的情况下）: 采用数据总量/机器数获取单机分页值，任务数量为机器数量