Elastic-JOB 定时任务

开发指南

代码开发

作业类型

3

Simple

DataFlow

Script

DataFlow

2

ThroughputDataFlow

SequenceDataFlow

Script

shardingContext

getShardingTotalCount()

getShardingItems()

Simple类型作业

Simple

AbstractSimpleElasticJob

Quartz

public class MyElasticJob extends AbstractSimpleElasticJob {

@Override
public void process(JobExecutionMultipleShardingContext context) {
// do something by sharding items
}
}

ThroughputDataFlow类型作业

ThroughputDataFlow

AbstractIndividualThroughputDataFlowElasticJob

3

fetchData

null

fetchData

processData

TbSchedule

fetchData

processData

processData

processData

fetchData

processData

false

true

public class MyElasticJob extends AbstractIndividualThroughputDataFlowElasticJob<Foo> {

@Override
public List<Foo> fetchData(JobExecutionMultipleShardingContext context) {
Map<Integer, String> offset = context.getOffsets();
List<Foo> result = // get data from database by sharding items and by offset
return result;
}

@Override
public boolean processData(JobExecutionMultipleShardingContext context, Foo data) {
// process data
// ...

// store offset
for (int each : context.getShardingItems()) {
updateOffset(each, "your offset, maybe id");
}
return true;
}
}

SequenceDataFlow类型作业

SequenceDataFlow

ThroughputDataFlow

ThroughputDataFlow

2

100

1

40

2

60

1

50

2

50

SequenceDataFlow

2

100

1

40

2

60

1

1

40

2

2

60

ThroughputDataFlow

SequenceDataFlow

public class MyElasticJob extends AbstractIndividualSequenceDataFlowElasticJob<Foo> {

@Override
public List<Foo> fetchData(JobExecutionSingleShardingContext context) {
int offset = context.getOffset();
List<Foo> result = // get data from database by sharding items and by offset
return result;
}

@Override
public boolean processData(JobExecutionSingleShardingContext context, Foo data) {
// process data
// ...

// store offset
updateOffset(context.getShardingItem(), "your offset, maybe id");
return true;
}
}

Script类型作业

Script

shell

python

perl

#!/bin/bash
echo sharding execution context is $*

sharding execution context is {"shardingItems":[0,1,2,3,4,5,6,7,8,9],"shardingItemParameters":{},"offsets":{},"jobName":"scriptElasticDemoJob","shardingTotalCount":10,"jobParameter":"","monitorExecution":true,"fetchDataCount":1}

批量处理

AbstractIndividualThroughputDataFlowElasticJob

AbstractIndividualSequenceDataFlowElasticJob

AbstractBatchThroughputDataFlowElasticJob

AbstractBatchSequenceDataFlowElasticJob

processData

boolean

int

List

异常处理

elastic-job

handleJobExecutionException

quartz

JobExecutionException

任务监听配置

每台作业节点均执行的监听

import com.dangdang.ddframe.job.api.JobExecutionMultipleShardingContext;
import com.dangdang.ddframe.job.api.listener.ElasticJobListener;

public class MyElasticJobListener implements ElasticJobListener {

@Override
public void beforeJobExecuted(final JobExecutionMultipleShardingContext shardingContext) {
// do something ...
}

@Override
public void afterJobExecuted(final JobExecutionMultipleShardingContext shardingContext) {
// do something ...
}
}

JobScheduler

public class JobMain {

public static void main(final String[] args) {
new JobScheduler(regCenter, jobConfig, new MyElasticJobListener()).init();
}
}

分布式场景中仅单一节点执行的监听

import com.dangdang.ddframe.job.api.JobExecutionMultipleShardingContext;
import com.dangdang.ddframe.job.api.listener.AbstractDistributeOnceElasticJobListener;

public final class TestDistributeOnceElasticJobListener extends AbstractDistributeOnceElasticJobListener {

public TestDistributeOnceElasticJobListener(final long startTimeoutMills, final long completeTimeoutMills) {
super(startTimeoutMills, completeTimeoutMills);
}

@Override
public void doBeforeJobExecutedAtLastStarted(final JobExecutionMultipleShardingContext shardingContext) {
// do something ...
}

@Override
public void doAfterJobExecutedAtLastCompleted(final JobExecutionMultipleShardingContext shardingContext) {
// do something ...
}
}

JobScheduler

public class JobMain {

public static void main(final String[] args) {
long startTimeoutMills = 5000L;
long completeTimeoutMills = 10000L;
new JobScheduler(regCenter, jobConfig, new MyDistributeOnceElasticJobListener(startTimeoutMills, completeTimeoutMills)).init();
}
}

作业配置

Spring

Bean

Spring
Bean

Spring

placeholder

Spring命名空间配置

<?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 ">
<!--配置作业注册中心 -->
<reg:zookeeper id="regCenter" server-lists=" yourhost:2181" namespace="dd-job" base-sleep-time-milliseconds="1000" max-sleep-time-milliseconds="3000" max-retries="3" />

<!-- 配置简单作业-->
<job:simple id="simpleElasticJob" class="xxx.MySimpleElasticJob" registry-center-ref="regCenter" cron="0/10 * * * * ?"   sharding-total-count="3" sharding-item-parameters="0=A,1=B,2=C" />

<!-- 配置数据流作业-->
<job:dataflow id="throughputDataFlow" class="xxx.MyThroughputDataFlowElasticJob" registry-center-ref="regCenter" cron="0/10 * * * * ?" sharding-total-count="3" sharding-item-parameters="0=A,1=B,2=C" process-count-interval-seconds="10" concurrent-data-process-thread-count="10" />

<!-- 配置脚本作业-->
<job:script id="scriptElasticJob" registry-center-ref="regCenter" cron="0/10 * * * * ?" sharding-total-count="3" sharding-item-parameters="0=A,1=B,2=C" script-command-line="/your/file/path/demo.sh" />

<!-- 配置带监听的简单作业-->
<job:simple id="listenerElasticJob" class="xxx.MySimpleListenerElasticJob" registry-center-ref="regCenter" cron="0/10 * * * * ?"   sharding-total-count="3" sharding-item-parameters="0=A,1=B,2=C">
<job:listener class="xx.MySimpleJobListener"/>
<job:listener class="xx.MyOnceSimpleJobListener" started-timeout-milliseconds="1000" completed-timeout-milliseconds="2000" />
</job:simple>
</beans>

job:simple命名空间属性详细说明

是

ElasticJob

是

Bean

reg:zookeeper

是

cron

是

0

0=a,1=b,2=c

-1

monitorExecution

job:dataflow命名空间属性详细说明

ThroughputDataFlow

fetchData

job:script命名空间属性详细说明，基本属性参照job:simple命名空间属性详细说明

job:listener命名空间属性详细说明

job:listener

job:bean

是

ElasticJobListener

否

否

reg:bean命名空间属性详细说明

是

Spring

是

Zookeeper

是

Zookeeper

Zookeeper

Zookeeper

Zookeeper

Zookeeper

Zookeeper

不使用Spring配置

import com.dangdang.ddframe.job.api.config.JobConfiguration;
import com.dangdang.ddframe.job.api.JobScheduler;
import com.dangdang.ddframe.reg.base.CoordinatorRegistryCenter;
import com.dangdang.ddframe.reg.zookeeper.ZookeeperConfiguration;
import com.dangdang.ddframe.reg.zookeeper.ZookeeperRegistryCenter;
import com.dangdang.example.elasticjob.core.job.SimpleJobDemo;
import com.dangdang.example.elasticjob.core.job.ThroughputDataFlowJobDemo;
import com.dangdang.example.elasticjob.core.job.SequenceDataFlowJobDemo;
import com.dangdang.ddframe.job.plugin.job.type.integrated.ScriptElasticJob;

public class JobDemo {

// 定义Zookeeper注册中心配置对象
private ZookeeperConfiguration zkConfig = new ZookeeperConfiguration("localhost:2181", "elastic-job-example", 1000, 3000, 3);

// 定义Zookeeper注册中心
private CoordinatorRegistryCenter regCenter = new ZookeeperRegistryCenter(zkConfig);

// 定义简单作业配置对象
private final SimpleJobConfiguration simpleJobConfig = JobConfigurationFactory.createSimpleJobConfigurationBuilder("simpleElasticDemoJob",
SimpleJobDemo.class, 10, "0/30 * * * * ?").build();

// 定义高吞吐流式处理的数据流作业配置对象
private final DataFlowJobConfiguration throughputJobConfig = JobConfigurationFactory.createDataFlowJobConfigurationBuilder("throughputDataFlowElasticDemoJob",
ThroughputDataFlowJobDemo.class, 10, "0/5 * * * * ?").streamingProcess(true).build();

// 定义顺序的数据流作业配置对象
private final DataFlowJobConfiguration sequenceJobConfig = JobConfigurationFactory.createDataFlowJobConfigurationBuilder("sequenceDataFlowElasticDemoJob",
SequenceDataFlowJobDemo.class, 10, "0/5 * * * * ?").build();

// 定义脚本作业配置对象
private final ScriptJobConfiguration scriptJobConfig = JobConfigurationFactory.createScriptJobConfigurationBuilder("scriptElasticDemoJob",
10, "0/5 * * * * ?", "test.sh").build();

public static void main(final String[] args) {
new JobDemo().init();
}

private void init() {
// 连接注册中心
regCenter.init();
// 启动简单作业
new JobScheduler(regCenter, simpleJobConfig).init();
// 启动高吞吐流式处理的数据流作业
new JobScheduler(regCenter, throughputJobConfig).init();
// 启动顺序的数据流作业
new JobScheduler(regCenter, sequenceJobConfig).init();
// 启动脚本作业
new JobScheduler(regCenter, scriptJobConfig).init();
}
}