企业级分布式批处理方案

@ComponentpublicclassParallelJobextendsMapReduceJobProcessor {
privatestaticfinalLoggerlogger=LoggerFactory.getLogger("schedulerx");
@OverridepublicProcessResultreduce(JobContextcontext) throwsException {
returnnewProcessResult(true);
    }
@OverridepublicProcessResultprocess(JobContextcontext) throwsException {
if(isRootTask(context)){
// ---------------------------------------------------------// Step1. 子任务的拆分逻辑开发，子任务对象可为业务自定义（内容务必简洁）// ---------------------------------------------------------logger.info("构建子任务列表...");
List<ParallelAccountInfo>list=newLinkedList();
/***  判断如果是rootTask的情况下，构建子任务对象列表*  在实际业务场景中，用户可自行根据业务场景加载子任务对象且该业务对象实现BizSubTask接口*  场景案例：*  1、从数据库中加载未被处理的客户账户信息*  2、构建省份城市地区信息列表，按区域分发任务处理*  3、根据业务标签作为子任务分类，如：电器、日用品、食品等*  4、可根据时间作为子任务分类，如：按月（1月、2月...）*/for(inti=0; i<20; i++){
list.add(newParallelAccountInfo(i, "CUS"+StringUtils.leftPad(i+"", 4, "0"),
"AC"+StringUtils.leftPad(i+"", 12, "0")));
            }
returnmap(list, "transfer");
        }else {
// ---------------------------------------------------------// Step2. 针对单个子任务的业务逻辑处理// ---------------------------------------------------------/*** 非rootTask，用户可以获取对应的子任务信息进行相应的业务处理*/ParallelAccountInfoobj= (ParallelAccountInfo)context.getTask();
// 针对获取的 obj子任务信息，强制还原为业务对象进行相应业务逻辑处理logger.info("处理子任务信息:{}", JSON.toJSONString(obj));
returnnewProcessResult(true);
        }
    }
}

publicclassParallelAccountInfoimplementsBizSubTask {
// 业务参数按业务场景需求自定义privatelongid;
privateStringname;
privateStringaccountId;
publicParallelAccountInfo(longid, Stringname, StringaccountId) {
this.id=id;
this.name=name;
this.accountId=accountId;
    }
/*** 自定义处理核心：* 业务实体需实现BizSubTask:labelMap方法，* 在该方法实现中设置对应子任务的标签信息，挑取子任务中核心业务信息作为lable* @return*/@OverridepublicMap<String, String>labelMap() {
Map<String, String>labelMap=newHashMap();
labelMap.put("户名", name);
returnlabelMap;
    }
}

@ComponentpublicclassFileImportJobextendsMapReduceJobProcessor {
privatestaticfinalLoggerlogger=LoggerFactory.getLogger("schedulerx");
@OverridepublicProcessResultreduce(JobContextcontext) throwsException {
returnnewProcessResult(true);
    }
@OverridepublicProcessResultprocess(JobContextcontext) throwsException {
if(isRootTask(context)){
// ---------------------------------------------------------// Step1. 读取对接的销售商列表Code// ---------------------------------------------------------logger.info("以销售商为维度构建子任务列表...");
// 伪代码从数据库读取销售商列表，Agency类需要实现BizSubTask接口并可将// 销售商名称/编码作为子任务标签，以便控制台可视化跟踪List<Agency>agencylist=getAgencyListFromDb();
returnmap(agencylist, "fileImport");
        }else {
// ---------------------------------------------------------// Step2. 针对单个销售商进行对应文件数据的处理// ---------------------------------------------------------Agencyagency= (Agency)context.getTask();
Filefile=loadFile(agency);
logger.info("文件加载完成.");
validFile(file);
logger.info("文件校验通过.");
List<Request>request=resolveRequest(file);
logger.info("文件数据解析完成.");
List<Request>request=checkRequest(request);
logger.info("申请数据检查通过.");
importRequest(request);
logger.info("申请数据导入完成.");
returnnewProcessResult(true);
        }
    }
}