elastic-job之监听器

淩亂°似流年 2022-06-02 12:36 348阅读 0赞

每个作业都可以配置一个任务监听器，确切的说是只能配置一个本地监听器和一个分布式监听器。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定义了一个本地监听器。

## 分布式监听器 ##

本地监听器在作业执行本地的分片任务时会执行，如上面的示例，我们的作业被分成了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/>`，以下是一个使用分布式监听器的示例：

（本文由Elim写于2017年10月2日）