elastic-job之流式作业

  1. 云栖社区>
  2. 博客>
  3. 正文

elastic-job之流式作业

elim1 2017-12-30 17:02:00 浏览1177
展开阅读全文

流式作业对应于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日)

网友评论

登录后评论
0/500
评论
elim1
+ 关注