廢話不多說了，直接上程式碼，elastic-job的一些配置引數可以參考這篇博文：https://blog.csdn.net/dhj199181/article/details/83088036

elastic-job.xml配置：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:reg="http://www.dangdang.com/schema/ddframe/reg"
    xmlns:job="http://www.dangdang.com/schema/ddframe/job"
    xsi:schemaLocation="http://www.springframework.org/schema/beans 
                        http://www.springframework.org/schema/beans/spring-beans.xsd 
                        http://www.dangdang.com/schema/ddframe/reg 
                        http://www.dangdang.com/schema/ddframe/reg/reg.xsd 
                        http://www.dangdang.com/schema/ddframe/job 
                        http://www.dangdang.com/schema/ddframe/job/job.xsd 
                        ">

    <!-- 配置作業註冊中心; baseSleepTimeMilliseconds:等待重試的間隔時間的初始值單位：毫秒 ; 
    maxSleepTimeMilliseconds:等待重試的間隔時間的最大值單位：毫秒;maxRetries:最大重試次數 -->
    <reg:zookeeper id="regCenter" server-lists="localhost:2181"
        namespace="elastic-job" base-sleep-time-milliseconds="1000"
        max-sleep-time-milliseconds="3000" max-retries="3" />
        
        <!-- 配置簡單作業 -->
    <job:simple id="JobSimpleJob" class="com.it.park.rms.job.MyElasticSimpleJob"
        registry-center-ref="regCenter" cron="0/30 * * * * ?"
        sharding-total-count="3" sharding-item-parameters="0=A,1=B,2=C" />
        
        <!-- 配置流式作業 -->
    <job:dataflow id="myDataFlowJob" class="com.it.park.rms.job.MyDataFlowJob" registry-center-ref="regCenter"
              sharding-total-count="2"  sharding-item-parameters="0=A,1=B,2=C,3=D,4=E"  description="分片處理" cron="0 0/1 * * * ?" streaming-process="true" overwrite="true" />
        
        
</beans>
 

SimpleJob任務例子：

public class MyElasticSimpleJob implements SimpleJob{

	@Override
	public void execute(ShardingContext arg0) {
		System.out.println("-------");
		
	}

}

DataflowJob任務例子：

public class MyDataFlowJob implements DataflowJob<TestModel>{

	@Autowired
	private TestService testService;
	
	@Override
	public List<TestModel> fetchData(ShardingContext arg0) {
		System.out.println(arg0.getJobName());
		System.out.println(arg0.getJobParameter());
		System.out.println(arg0.getShardingItem());
		System.out.println(arg0.getShardingTotalCount());
		System.out.println(arg0.getTaskId());
		TestModel model = new TestModel();
		model.setStatus("N");
		List<TestModel> list = testService.selectByModel(model);
		System.out.println("======"+list.size());
		return list;
	}

	@Override
	public void processData(ShardingContext arg0, List<TestModel> list) {
		for (TestModel testModel : list) {
			System.out.println(testModel.getName());
			testService.batchUpdate(Lists.newArrayList(testModel));
			System.out.println("修改成功");
		}
	}

}