/**
 * 弹性化分布式作业监听器接口.
 *
 * @author zhangliang
 */
public interface ElasticJobListener {/**
     * 作业执行前的执行的方法.
     *
     * @param shardingContexts 分片上下文
     */void beforeJobExecuted(final ShardingContexts shardingContexts);/**
     * 作业执行后的执行的方法.
     *
     * @param shardingContexts 分片上下文
     */void afterJobExecuted(final ShardingContexts shardingContexts);
}

public class MyElasticJobListener implements ElasticJobListener {private static final Logger LOGGER = Logger.getLogger(MyElasticJobListener.class);@Overridepublic void beforeJobExecuted(ShardingContexts shardingContexts) {LOGGER.info(String.format("开始调度任务[%s]", shardingContexts.getJobName()));}@Overridepublic void afterJobExecuted(ShardingContexts shardingContexts) {LOGGER.info(String.format("任务[%s]调度完成", shardingContexts.getJobName()));}}

<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>

/**
 * 在分布式作业中只执行一次的监听器.
 *
 * @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();@Setterprivate 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;}}@Overridepublic 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);}}@Overridepublic 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);}@Overridepublic void doBeforeJobExecutedAtLastStarted(ShardingContexts shardingContexts) {logger.info("分布式监听器开始……");}@Overridepublic void doAfterJobExecutedAtLastCompleted(ShardingContexts shardingContexts) {logger.info("分布式监听器结束……");}}

<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>