elastic-job之流式作业

我就是我 2022-06-02 02:23 214阅读 0赞

流式作业对应于DataflowJob接口，其定义如下：

/** * 数据流分布式作业接口. * * @author zhangliang * * @param <T> 数据类型 */
    public interface DataflowJob<T> extends ElasticJob { 
    
        /** * 获取待处理数据. * * @param shardingContext 分片上下文 * @return 待处理的数据集合 */
        List<T> fetchData(ShardingContext shardingContext);
    
        /** * 处理数据. * * @param shardingContext 分片上下文 * @param data 待处理数据集合 */
        void processData(ShardingContext shardingContext, List<T> data);
    }

流式作业，每次调度触发的时候都会先调fetchData获取数据，如果获取到了数据再调度processData方法处理数据。DataflowJob在运行时有两种方式，流式的和非流式的，通过属性streamingProcess控制，如果是基于Spring XML的配置方式则是streaming-process属性，boolean类型。当作业配置为流式的时候，每次触发作业后会调度一次fetchData获取数据，如果获取到了数据会调度processData方法处理数据，处理完后又继续调fetchData获取数据，再调processData处理，如此循环，就像流水一样。直到fetchData没有获取到数据或者发生了重新分片才会停止。代码实现部分可参考数据流执行器 com.dangdang.ddframe.job.executor.type.DataflowJobExecutor。以下是DataflowJob的一个简单实现，该实现中每次调度触发时都会连续调度processData十次。

public class MyDataflowJob implements DataflowJob<String> { 
    
        private static final ThreadLocal<Integer> LOOP_COUNTER = new ThreadLocal<>();
        private static final int LOOP_TIMES = 10;//每次获取流处理循环次数
        private static final AtomicInteger COUNTER = new AtomicInteger(1);//计数器
    
        @Override
        public List<String> fetchData(ShardingContext shardingContext) {
            Integer current = LOOP_COUNTER.get();
            if (current == null) {
                current = 1;
            } else {
                current += 1;
            }
            LOOP_COUNTER.set(current);
            System.out.println(Thread.currentThread() + "------------current--------" + current);
            if (current > LOOP_TIMES) {
                System.out.println("\n\n\n\n");
                return null;
            } else {
                int shardingItem = shardingContext.getShardingItem();
                List<String> datas = Arrays.asList(getData(shardingItem), getData(shardingItem), getData(shardingItem));
                return datas;
            }
        }
    
        private String getData(int shardingItem) {
            return shardingItem + "-" + COUNTER.getAndIncrement();
        }
    
        @Override
        public void processData(ShardingContext shardingContext, List<String> data) {
            System.out.println(Thread.currentThread() + "--------" +data);
        }
    
    }

流式作业的配置使用`<job:dataflow/>`配置，上面的流式作业对应的配置如下：

<job:dataflow id="myDataflowJob" class="com.elim.learn.elastic.job.MyDataflowJob" registry-center-ref="regCenter" cron="0 0/2 * * * ?" sharding-total-count="2" sharding-item-parameters="0=广州,1=深圳" failover="true" overwrite="true" streaming-process="true">

上述配置参数的含义跟上一篇介绍的简单作业的配置是一样的，新增的streaming-process表示是否启用流式作业。

（本文由Elim写于2017年10月1日）