基于任务调度的企业级分布式批处理方案

elasticjob:  regCenter:    serverLists: 127.0.0.1:2181    namespace: elasticjob-lite-springboot  jobs:    simpleJob:      elasticJobClass: org.example.job.SpringBootSimpleJob      cron: 0/5 * * * * ?      overwrite: true      shardingTotalCount: 3      shardingItemParameters: 0=Beijing,1=Shanghai,2=Guangzhou

@Componentpublic class SpringBootSimpleJob implements SimpleJob {    @Override    public void execute(final ShardingContext shardingContext) {        String value = shardingContext.getShardingParameter();        System.out.println("simple.process->"+value);    }}

public interface DataflowJob<T> extends ElasticJob {
    /**     * Fetch to be processed data.     *     * @param shardingContext sharding context     * @return to be processed data     */    List<T> fetchData(ShardingContext shardingContext);
    /**     * Process data.     *     * @param shardingContext sharding context     * @param data to be processed data     */    void processData(ShardingContext shardingContext, List<T> data);}

elasticjob:  regCenter:    serverLists: 127.0.0.1:2181    namespace: elasticjob-lite-springboot  jobs:    dataflowJob:      elasticJobClass: org.example.job.SpringBootDataflowJob      cron: 0/5 * * * * ?      overwrite: true      shardingTotalCount: 3      shardingItemParameters: 0=Beijing,1=Shanghai,2=Guangzhou      props:        # 开启streaming process        streaming.process: true

    @Bean    public Job remoteChunkingJob() {         return jobBuilderFactory.get("remoteChunkingJob")             .start(stepBuilderFactory.get("step2")                     .<Integer, Integer>chunk(2) // 每Chunk块包含reader加载的记录数                     .reader(itemReader())                     // 采用ChunkMessageChannelItemWriter分发Chunk块                     .writer(itemWriter())                     .build())             .build();     }
    @Bean    public ItemReader<Integer> itemReader() {        return new ListItemReader<>(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));    }
    @Bean    public ItemWriter<Integer> itemWriter() {        MessagingTemplate messagingTemplate = new MessagingTemplate();        messagingTemplate.setDefaultChannel(requests());        ChunkMessageChannelItemWriter<Integer> chunkMessageChannelItemWriter = new ChunkMessageChannelItemWriter<>();        chunkMessageChannelItemWriter.setMessagingOperations(messagingTemplate);        chunkMessageChannelItemWriter.setReplyChannel(replies());        return chunkMessageChannelItemWriter;    }    // 省略了相关消息中间件对接通道配置

    // 省略了相关消息中间件对接通道配置
    // 接收分块任务升级及反馈执行结果    @Bean    @ServiceActivator(inputChannel = "slaveRequests", outputChannel = "slaveReplies")    public ChunkProcessorChunkHandler<Integer> chunkProcessorChunkHandler() {        ChunkProcessor<Integer> chunkProcessor = new SimpleChunkProcessor(slaveItemProcessor(), slaveItemWriter());        ChunkProcessorChunkHandler<Integer> chunkProcessorChunkHandler = new ChunkProcessorChunkHandler<>();        chunkProcessorChunkHandler.setChunkProcessor(chunkProcessor);        return chunkProcessorChunkHandler;    }
    // 实际业务需要开发的任务处理逻辑processor    @Bean    public SlaveItemProcessor slaveItemProcessor(){ return new SlaveItemProcessor();}
    // 实际业务需要开发的任务处理逻辑writer    @Bean    public SlaveItemWriter slaveItemWriter(){ return new SlaveItemWriter();}

public class MapReduceHelloProcessor extends MapReduceJobProcessor {
    @Override    public ProcessResult reduce(JobContext jobContext) throws Exception {        // 所有子任务完成的汇聚逻辑处理回调，可选实现        jobContext.getTaskResults();        return new ProcessResult(true, "处理结果数量集：" + jobContext.getTaskResults().size());    }
    @Override    public ProcessResult process(JobContext jobContext) throws Exception {        if (isRootTask(jobContext)) {            List<String> list = // 加载业务待处理的子任务列表            // 回调sdk提供的map方法，自动实现子任务分发            ProcessResult result = map(list, "SecondDataProcess");            return result;        } else {            // 获得单个子任务数据信息，进行单个子任务业务处理            String data = (String) jobContext.getTask();            // ... 业务逻辑处理补充 ...            return new ProcessResult(true, "数据处理成功！");        }    }}

public class AccountTransferProcessor extends MapReduceJobProcessor {
    private static final Logger logger = LoggerFactory.getLogger("schedulerxLog");
    @Override    public ProcessResult reduce(JobContext context) throws Exception {        return new ProcessResult(true);    }
    @Override    public ProcessResult process(JobContext context) throws Exception {        if(isRootTask(context)){            logger.info("split task list size:20");            List<AccountInfo> list = new LinkedList();            for(int i=0; i < 20; i++){                list.add(new AccountInfo(i, "CUS"+StringUtils.leftPad(i+"", 4, "0"),                        "AC"+StringUtils.leftPad(i+"", 12, "0")));            }            return map(list, "transfer");        }else {            logger.info("start biz process...");            logger.info("task info:"+context.getTask().toString());            TimeUnit.SECONDS.sleep(30L);            logger.info("start biz process end.");            return new ProcessResult(true);        }    }}
public class AccountInfo implements BizSubTask {
        private long id;
        private String name;
        private String accountId;
        public AccountInfo(long id, String name, String accountId) {            this.id = id;            this.name = name;            this.accountId = accountId;        }
        // 子任务标签信息设置        @Override        public Map<String, String> labelMap() {            Map<String, String> labelMap = new HashMap();            labelMap.put("账户名称", name);            return labelMap;        }    }

@Componentpublic class FileImportJob extends MapReduceJobProcessor {
    private static final Logger logger = LoggerFactory.getLogger("schedulerx");
    @Override    public ProcessResult reduce(JobContext context) throws Exception {        return new ProcessResult(true);    }
    @Override    public ProcessResult process(JobContext context) throws Exception {        if(isRootTask(context)){            // ---------------------------------------------------------            // Step1. 读取对接的销售商列表Code            // ---------------------------------------------------------            logger.info("以销售商为维度构建子任务列表...");
            // 伪代码从数据库读取销售商列表，Agency类需要实现BizSubTask接口并可将            // 销售商名称/编码作为子任务标签，以便控制台可视化跟踪            List<Agency> agencylist = getAgencyListFromDb();            return map(agencylist, "fileImport");        }else {            // ---------------------------------------------------------            // Step2. 针对单个销售商进行对应文件数据的处理            // ---------------------------------------------------------            Agency agency = (Agency)context.getTask();            File file = 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("申请数据导入完成.");            return new ProcessResult(true);        }    }}