源码分析ElasticJob任务运行事件监听器

  1. 云栖社区>
  2. 中间件兴趣圈>
  3. 博客>
  4. 正文

源码分析ElasticJob任务运行事件监听器

丁威 2019-10-08 21:18:29 浏览2586
展开阅读全文

在任务执行的前后,ElasticJob可以提供扩展,其主要类图如下:
这里写图片描述
ElastciJobListener:elasticJob任务执行事件监听器,提供如下两个方法:

  • void beforeJobExecuted(final ShardingContexts shardingContexts);

     在任务执行之前调用,shardingContexts为分片上下文信息。
  • void afterJobExecuted(final ShardingContexts shardingContexts)

    在任务执行之后调用,shardingContexts为分片上下文信息。
    

上述回调函数是分片级的,也就是说默认情况下,同一个任务的多个分片都会执行beforeJobExecuted、afterJobExecuted方法,如果某些情况同一个任务只需在最后一个分片执行之前执行,最后一个分片执行完成后才执行,又该如何实现呢。AbstractDistributeOnceElasticJobListener粉墨登场。

AbstractDistributeOnceElasticJobListener:在分布式作业中只执行一次的监听器。

  • private final long startedTimeoutMilliseconds:分片等待beforeJobExecuted方法执行的超时时间,单位为毫秒。
  • private final Object startedWait = new Object():分片等待beforeJobExecuted的监视器。
  • private final long completedTimeoutMilliseconds:分片等待afterJobExecuted方法执行的超时时间,单位为毫秒。
  • private final Object completedWait = new Object():分片等待afterJobExecuted的监视器。
  • private GuaranteeService guaranteeService:保证分布式任务全部开始和结束状态的服务。
  • private TimeService timeService = new TimeService():时间服务器,主要用来获取当前服务器的系统时间。
  • public final void beforeJobExecuted(final ShardingContexts shardingContexts):分片任务执行之前调用,该方法是一个模板方法,最后一个分片成功启动后调用doBeforeJobExecutedAtLastStarted方法,该方法为抽象方法,由具体子类实现,如果有其他分片未执行完成,该方法会阻塞等待,或最后启动的分片执行完doBeforeJobExecutedAtLastStarted方法。
  • public final void afterJobExecuted(final ShardingContexts shardingContexts):分片任务执行之后调用,该方法是一个模板方法,实现当最后一个分片成功执行完成后调用doAfterJobExecutedAtLastCompleted方法,该方法为抽象方法,由具体子类实现,如果有其他分片未执行完成,该方法会阻塞等待,或最后启动的分片执行完doAfterJobExecutedAtLastCompleted方法。
  • public abstract void doBeforeJobExecutedAtLastStarted(ShardingContexts shardingContexts):分布式环境中最后一个作业分片执行前的执行的方法。
  • public abstract void doAfterJobExecutedAtLastCompleted(ShardingContexts shardingContexts):分布式环境中最后一个作业分片执行完成后的执行方法。
  • public void notifyWaitingTaskStart():通知分片节点上的任务开始之前(唤醒由于还有其他分片未启动造成自身等待阻塞)。
  • public void notifyWaitingTaskComplete():通知分片节点任务执行完成(唤醒由于存在其他分片任务未执行完成时阻塞)。

接下来重点分析AbstractDistributeOnceElasticJobListener实现原理(分布式环境中,监听器只在一个节点上执行的实现逻辑)

重点分析beforeJobExecuted方法实现原理,afterJobExecuted方法类似。

AbstractDistributeOnceElasticJobListener#beforeJobExecuted

public final void beforeJobExecuted(final ShardingContexts shardingContexts) {
        guaranteeService.registerStart(shardingContexts.getShardingItemParameters().keySet());    // @1
        if (guaranteeService.isAllStarted()) {                                                                                         // @2
            doBeforeJobExecutedAtLastStarted(shardingContexts);
            guaranteeService.clearAllStartedInfo();
            return;
        }
        long before = timeService.getCurrentMillis();                                                                           // @3
        try {
            synchronized (startedWait) {
                startedWait.wait(startedTimeoutMilliseconds);
            }
        } catch (final InterruptedException ex) {
            Thread.interrupted();
        }
        if (timeService.getCurrentMillis() - before >= startedTimeoutMilliseconds) {                         // @4
            guaranteeService.clearAllStartedInfo();
            handleTimeout(startedTimeoutMilliseconds);
        }
    }

代码@1:使用GuaranteeService注册分片开始。

代码@2:判断该任务所有的分片是否都已经注册启动,如果都注册启动,则调用doBeforeJobExecutedAtLastStarted()方法。

代码@3:获取服务器当前时间。
代码@4:利用startWait.wait(startedTimeoutMilliseconds)带超时时间的等待,这里如何唤醒呢?
代码@5:判断唤醒是超时唤醒还是正常唤醒,如果是超时唤醒,清除所有的分片注册启动信息,处理超时异常。

上述流程简单明了,上面有两个问题需要进一步探究,如何注册分片启动信息与如何被唤醒。

1、任务节点注册分配给当前节点的任务分片

/**
     * 根据分片项注册任务开始运行.
     * 
     * @param shardingItems 待注册的分片项
     */
    public void registerStart(final Collection<Integer> shardingItems) {
        for (int each : shardingItems) {
            jobNodeStorage.createJobNodeIfNeeded(GuaranteeNode.getStartedNode(each));
        }
    }

创建持久节点:${namespace}/jobname/guarantee/started/{item}。

2、当最后一个节点注册启动执行doBeforeJobExecutedAtLastStarted方法后,如果唤醒其他节点以便进入到任务执行阶段

if (guaranteeService.isAllStarted()) {
        doBeforeJobExecutedAtLastStarted(shardingContexts);
        guaranteeService.clearAllStartedInfo();
        return;
 }

也就是回调函数执行完毕后,会删除任务所有的分片。温馨提示:业务实现子类实现doBeforeJobExecutedAtLastStarted方法时最好不要抛出异常,不然各节点的唤醒操作只能是等待超时后被唤醒。

GuaranteeService#clearAllStartedInfo

/**
     * 清理所有任务启动信息.
     */
    public void clearAllStartedInfo() {
        jobNodeStorage.removeJobNodeIfExisted(GuaranteeNode.STARTED_ROOT);
    }

直接删除${namespace}/jobname/guarantee/started根节点。基于ZK的开发模式,触发一次删除操作,肯定会有事件监听器来监听该节点的删除事件,从而触发其他节点的唤醒操作,果不奇然,ElastciJob提供GuaranteeListenerManager事件监听来监听${namespace}/jobname/guarantee/started节点的删除事件。

GuaranteeListenerManager$StartedNodeRemovedJobListener

class StartedNodeRemovedJobListener extends AbstractJobListener {
        @Override
        protected void dataChanged(final String path, final Type eventType, final String data) {
            if (Type.NODE_REMOVED == eventType && guaranteeNode.isStartedRootNode(path)) {
                for (ElasticJobListener each : elasticJobListeners) {
                    if (each instanceof AbstractDistributeOnceElasticJobListener) {
                        ((AbstractDistributeOnceElasticJobListener) each).notifyWaitingTaskStart();
                    }
                }
            }
        }
    }

每个Job实例在监听到${namespace}/jobname/guarantee/started节点被删除后,会执行AbstractDistributeOnceElasticJobListener的notifyWaitingTaskStart方法唤醒被阻塞的线程,是线程进入到任务执行阶段。

同理,任务执行后监听方法afterJobExecuted的执行流程实现原理一样,在这里就不在重复讲解了。


原文发布时间为:2018-12-15
本文作者:丁威,《RocketMQ技术内幕》作者。
本文来自中间件兴趣圈,了解相关信息可以关注中间件兴趣圈

网友评论

登录后评论
0/500
评论
丁威
+ 关注
所属云栖号: 中间件兴趣圈