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