【非广告】分布式调度框架 elastic-job 实践详解（超详细）（三）

@Slf4j
public class MyDataFlowJob implements DataflowJob<String> {
    private boolean flag = false;
    @Override
    public List<String> fetchData(ShardingContext shardingContext) {
        log.info("开始获取数据");
        if (flag) {
            return null;
        }
        return Arrays.asList("qingshan", "jack", "seven");
    }
    @Override
    public void processData(ShardingContext shardingContext, List<String> data) {
        for (String val : data) {
            // 处理完数据要移除掉，不然就会一直跑,处理可以在上面的方法里执行。这里采用 flag
            log.info("开始处理数据：" + val);
        }
        flag = true;
    }
}

Configuration
public class MyDataFlowJobConfig {
    /**
     * 任务名称
     */
    @Value("${dataflowJob.myDataflowJob.name}")
    private String jobName;
    /**
     * cron表达式
     */
    @Value("${dataflowJob.myDataflowJob.cron}")
    private String jobCron;
    /**
     * 作业分片总数
     */
    @Value("${dataflowJob.myDataflowJob.shardingTotalCount}")
    private int jobShardingTotalCount;
    /**
     * 作业分片参数
     */
    @Value("${dataflowJob.myDataflowJob.shardingItemParameters}")
    private String jobShardingItemParameters;
    /**
     * 自定义参数
     */
    @Value("${dataflowJob.myDataflowJob.jobParameters}")
    private String jobParameters;
    @Autowired
    private ZookeeperRegistryCenter registryCenter;
    @Bean
    public MyDataFlowJob myDataFlowJob() {
        return new MyDataFlowJob();
    }
    @Bean(initMethod = "init")
    public JobScheduler dataFlowJobScheduler(final MyDataFlowJob myDataFlowJob) {
        MyElasticJobListener elasticJobListener = new MyElasticJobListener();
        return new SpringJobScheduler(myDataFlowJob, registryCenter, getLiteJobConfiguration(), elasticJobListener);
    }
    private LiteJobConfiguration getLiteJobConfiguration() {
        // 定义作业核心配置
        JobCoreConfiguration dataflowCoreConfig = JobCoreConfiguration.newBuilder(jobName, jobCron, jobShardingTotalCount).
                shardingItemParameters(jobShardingItemParameters).jobParameter(jobParameters).build();
        // 定义DATAFLOW类型配置
        DataflowJobConfiguration dataflowJobConfig = new DataflowJobConfiguration(dataflowCoreConfig, MyDataFlowJob.class.getCanonicalName(), false);
        // 定义Lite作业根配置
        LiteJobConfiguration dataflowJobRootConfig = LiteJobConfiguration.newBuilder(dataflowJobConfig).overwrite(true).build();
        return dataflowJobRootConfig;
    }
}

#dataflow类型的job
dataflowJob.myDataflowJob.name=myDataflowJob
dataflowJob.myDataflowJob.cron=0/15 * * * * ?
dataflowJob.myDataflowJob.shardingTotalCount=1
dataflowJob.myDataflowJob.shardingItemParameters=0=a,1=b,2=c
dataflowJob.myDataflowJob.jobParameters=myDataflowJobParamter