每个作业都可以配置一个任务监听器，确切的说是只能配置一个本地监听器和一个分布式监听器。Elastic-job有三种作业类型，但是它们的通用配置都是一样的，所以本文在介绍作业的监听器配置时将仅以简单作业的配置为例。
本地监听器
本地监听器只在节点执行自己分片的时候调度，每个分片任务调度的时候本地监听器都会执行。本地监听器由ElasticJobListener接口定义，其定义如下：
/**
* 弹性化分布式作业监听器接口.
*
* @author zhangliang
*/
public interface ElasticJobListener {
/**
* 作业执行前的执行的方法.
*
* @param shardingContexts 分片上下文
*/
void beforeJobExecuted(final ShardingContexts shardingContexts);
/**
* 作业执行后的执行的方法.
*
* @param shardingContexts 分片上下文
*/
void afterJobExecuted(final ShardingContexts shardingContexts);
}
该接口的接口方法的注释上已经说明了对应的接口方法的调用时机，详情也可以参考com.dangdang.ddframe.job.executor.AbstractElasticJobExecutor.execute()方法。简单示例如下：
public class MyElasticJobListener implements ElasticJobListener {
private static final Logger LOGGER = Logger.getLogger(MyElasticJobListener.class);
@Override
public void beforeJobExecuted(ShardingContexts shardingContexts) {
LOGGER.info(String.format("开始调度任务[%s]", shardingContexts.getJobName()));
}
@Override
public void afterJobExecuted(ShardingContexts shardingContexts) {
LOGGER.info(String.format("任务[%s]调度完成", shardingContexts.getJobName()));
}
}
本地监听器的配置由<job:listener/>节点配置，如下示例中就通过<job:listener/>给简单作业myElasticJob定义了一个本地监听器。
<bean id="simpleJob" class="com.elim.learn.elastic.job.MyElasticJob"/>
<job:simple id="myElasticJob" job-ref="simpleJob"
registry-center-ref="regCenter" cron="0/30 * * * * ?"
sharding-total-count="6" sharding-item-parameters="0=A,1=B,2=C,3=D,4=E,5=F"
failover="true" overwrite="true" >
<job:listener class="com.elim.learn.elastic.job.listener.MyElasticJobListener" />
</job:simple>
分布式监听器
本地监听器在作业执行本地的分片任务时会执行，如上面的示例，我们的作业被分成了6片，则监听器任务会执行6次。而分布式监听器会在总的任务开始执行时执行一次，在总的任务结束执行时执行一次。分布式监听器也是在普通监听器的基础上实现的，由AbstractDistributeOnceElasticJobListener抽象类封装的，其实现了ElasticJobListener接口。要实现自己的监听器只需要继承AbstractDistributeOnceElasticJobListener抽象类，实现其中的抽象方法即可。AbstractDistributeOnceElasticJobListener抽象类的定义如下：
/**
* 在分布式作业中只执行一次的监听器.
*
* @author zhangliang
*/
public abstract class AbstractDistributeOnceElasticJobListener implements ElasticJobListener {
private final long startedTimeoutMilliseconds;
private final Object startedWait = new Object();
private final long completedTimeoutMilliseconds;
private final Object completedWait = new Object();
@Setter
private GuaranteeService guaranteeService;
private TimeService timeService = new TimeService();
public AbstractDistributeOnceElasticJobListener(final long startedTimeoutMilliseconds, final long completedTimeoutMilliseconds) {
if (startedTimeoutMilliseconds <= 0L) {
this.startedTimeoutMilliseconds = Long.MAX_VALUE;
} else {
this.startedTimeoutMilliseconds = startedTimeoutMilliseconds;
}
if (completedTimeoutMilliseconds <= 0L) {
this.completedTimeoutMilliseconds = Long.MAX_VALUE;
} else {
this.completedTimeoutMilliseconds = completedTimeoutMilliseconds;
}
}
@Override
public final void beforeJobExecuted(final ShardingContexts shardingContexts) {
guaranteeService.registerStart(shardingContexts.getShardingItemParameters().keySet());
if (guaranteeService.isAllStarted()) {
doBeforeJobExecutedAtLastStarted(shardingContexts);
guaranteeService.clearAllStartedInfo();
return;
}
long before = timeService.getCurrentMillis();
try {
synchronized (startedWait) {
startedWait.wait(startedTimeoutMilliseconds);
}
} catch (final InterruptedException ex) {
Thread.interrupted();
}
if (timeService.getCurrentMillis() - before >= startedTimeoutMilliseconds) {
guaranteeService.clearAllStartedInfo();
handleTimeout(startedTimeoutMilliseconds);
}
}
@Override
public final void afterJobExecuted(final ShardingContexts shardingContexts) {
guaranteeService.registerComplete(shardingContexts.getShardingItemParameters().keySet());
if (guaranteeService.isAllCompleted()) {
doAfterJobExecutedAtLastCompleted(shardingContexts);
guaranteeService.clearAllCompletedInfo();
return;
}
long before = timeService.getCurrentMillis();
try {
synchronized (completedWait) {
completedWait.wait(completedTimeoutMilliseconds);
}
} catch (final InterruptedException ex) {
Thread.interrupted();
}
if (timeService.getCurrentMillis() - before >= completedTimeoutMilliseconds) {
guaranteeService.clearAllCompletedInfo();
handleTimeout(completedTimeoutMilliseconds);
}
}
private void handleTimeout(final long timeoutMilliseconds) {
throw new JobSystemException("Job timeout. timeout mills is %s.", timeoutMilliseconds);
}
/**
* 分布式环境中最后一个作业执行前的执行的方法.
*
* @param shardingContexts 分片上下文
*/
public abstract void doBeforeJobExecutedAtLastStarted(ShardingContexts shardingContexts);
/**
* 分布式环境中最后一个作业执行后的执行的方法.
*
* @param shardingContexts 分片上下文
*/
public abstract void doAfterJobExecutedAtLastCompleted(ShardingContexts shardingContexts);
/**
* 通知任务开始.
*/
public void notifyWaitingTaskStart() {
synchronized (startedWait) {
startedWait.notifyAll();
}
}
/**
* 通知任务结束.
*/
public void notifyWaitingTaskComplete() {
synchronized (completedWait) {
completedWait.notifyAll();
}
}
}
以下是一个使用分布式监听器的示例：
public class MyDistributeOnceElasticJobListener extends AbstractDistributeOnceElasticJobListener {
private static final Logger logger = Logger.getLogger(MyDistributeOnceElasticJobListener.class);
/**
* @param startedTimeoutMilliseconds
* @param completedTimeoutMilliseconds
*/
public MyDistributeOnceElasticJobListener(long startedTimeoutMilliseconds, long completedTimeoutMilliseconds) {
super(startedTimeoutMilliseconds, completedTimeoutMilliseconds);
}
@Override
public void doBeforeJobExecutedAtLastStarted(ShardingContexts shardingContexts) {
logger.info("分布式监听器开始……");
}
@Override
public void doAfterJobExecutedAtLastCompleted(ShardingContexts shardingContexts) {
logger.info("分布式监听器结束……");
}
}
分布式监听器用到了锁的等待和通知，startedTimeoutMilliseconds和completedTimeoutMilliseconds分别用来指定作业开始前和完成后的对应的锁等待最大超时时间。分布式监听器由<job:distributed-listener/>，以下是一个使用分布式监听器的示例：
<bean id="simpleJob" class="com.elim.learn.elastic.job.MyElasticJob"/>
<job:simple id="myElasticJob" job-ref="simpleJob"
registry-center-ref="regCenter" cron="0/30 * * * * ?"
sharding-total-count="6" sharding-item-parameters="0=A,1=B,2=C,3=D,4=E,5=F"
failover="true" overwrite="true" >
<job:distributed-listener class="com.elim.learn.elastic.job.listener.MyDistributeOnceElasticJobListener"
started-timeout-milliseconds="100" completed-timeout-milliseconds="100"/>
</job:simple>
（本文由Elim写于2017年10月2日）
