[Elastic-Job source code analysis] - job listener

概述

参考文档：芋道源码-Elastic Job Lite
下载源码地址：Elastic-Job源码地址
Elastic-Job官网地址：【Elastic-Job官网文档地址】
Elastic-Job代码版本：3.0.0-RC1-SNAPSHOT

---

内容

1. 作业监听器uml图【插播idea的工具[Diagrams],It is very helpful to analyze the source code,建议使用】
   

   1. ElasticJobListener,Each job node executes,aka local listener      
   2. AbstractDistributeOnceElasticJobListener实现了ElasticJobListener,重写beforeJobExecuted和afterJobExecuted,Only a single node executes in a distributed scenario
   3. AbstractDistributeOnceElasticJobListener依赖GuaranteeService类对zk节点的操作,Implement single node operation,They are combinatorial relationships
   4. AbstractListenerManageris the listener management interface,Defines a method for adding listener classes
   5. GuaranteeListenerManager实现了AbstractListenerManager,It is instantiated during initializationStartedNodeRemovedJobListener和CompletedNodeRemovedJobListener,Two classes listen for node changes,达到控制AbstractDistributeOnceElasticJobListenerWake-up of pre-listening and post-listening instances
   

2. ElasticJobListener
   There are mainly two methods in the source code：beforeJobExecuted和afterJobExecuted,If the job processes files from the job server,处理完成后删除文件,可考虑使用每个节点均执行清理任务.This type of task is simple to implement and does not need to consider whether the globally distributed task is completed,Please try to use this type of listener

   public interface ElasticJobListener {
         
   
   /** * 作业执行前的执行的方法. * * @param shardingContexts 分片上下文 */
   void beforeJobExecuted(final ShardingContexts shardingContexts);
   
   /** * 作业执行后的执行的方法. * * @param shardingContexts 分片上下文 */
   void afterJobExecuted(final ShardingContexts shardingContexts);}
   
   
   

• AbstractDistributeOnceElasticJobListener
  该接口实现ElasticJobListener,主要是重写了beforeJobExecuted和afterJobExecuted方法,Only let one of the multiple nodes execute.Depends on this implementationGuaranteeService功能,Implement only single node execution in distributed scenarios

  public abstract class AbstractDistributeOnceElasticJobListener implements ElasticJobListener {
        
  /** * Start the timeout period,Avoid deadlocks caused by unsynchronized jobs */
  private final long startedTimeoutMilliseconds;
  /** * Start waiting for the object,Avoid deadlocks caused by unsynchronized jobs */
  private final Object startedWait = new Object();
  /** * 完成超时时间 */
  private final long completedTimeoutMilliseconds;
  /** * Finish waiting for the object */
  private final Object completedWait = new Object();
  /** * A service that guarantees the start and end states of all distributed tasks */
  @Setter
  private GuaranteeService guaranteeService;
  
  private final TimeService timeService = new TimeService();
  
  public AbstractDistributeOnceElasticJobListener(final long startedTimeoutMilliseconds, final long completedTimeoutMilliseconds) {
        
      this.startedTimeoutMilliseconds = startedTimeoutMilliseconds <= 0L ? Long.MAX_VALUE : startedTimeoutMilliseconds;
      this.completedTimeoutMilliseconds = completedTimeoutMilliseconds <= 0L ? Long.MAX_VALUE : completedTimeoutMilliseconds;
  }
  
  @Override
  
  public final void beforeJobExecuted(final ShardingContexts shardingContexts) {
        
      Set<Integer> shardingItems = shardingContexts.getShardingItemParameters().keySet();
      if (shardingItems.isEmpty()) {
        
          return;
      }
      // The registration job shard entry starts running
      guaranteeService.registerStart(shardingItems);
      // Determine if all shard items are running,Not all are running,本线程休眠100ms
      while (!guaranteeService.isRegisterStartSuccess(shardingItems)) {
        
          BlockUtils.waitingShortTime();
      }
      // Determine if all shard items are running,开始线程
      if (guaranteeService.isAllStarted()) {
        
          // 执行
          doBeforeJobExecutedAtLastStarted(shardingContexts);
          // Clear boot information
          guaranteeService.clearAllStartedInfo();
          return;
      }
      long before = timeService.getCurrentMillis();
      try {
        
          synchronized (startedWait) {
        
              startedWait.wait(startedTimeoutMilliseconds);
          }
      } catch (final InterruptedException ex) {
        
          Thread.interrupted();
      }
  
      if (timeService.getCurrentMillis() - before >= startedTimeoutMilliseconds) {
        
          // Clear boot information
          guaranteeService.clearAllStartedInfo();
          // 抛出异常信息
          handleTimeout(startedTimeoutMilliseconds);
      }
  }
  
  @Override
  public final void afterJobExecuted(final ShardingContexts shardingContexts) {
        
      Set<Integer> shardingItems = shardingContexts.getShardingItemParameters().keySet();
      if (shardingItems.isEmpty()) {
        
          return;
      }
      guaranteeService.registerComplete(shardingItems);
      while (!guaranteeService.isRegisterCompleteSuccess(shardingItems)) {
        
          BlockUtils.waitingShortTime();
      }
      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);
  }
  
  /** * Do before job executed at last sharding job started. * * @param shardingContexts sharding contexts */
  public abstract void doBeforeJobExecutedAtLastStarted(ShardingContexts shardingContexts);
  
  /** * Do after job executed at last sharding job completed. * * @param shardingContexts sharding contexts */
  public abstract void doAfterJobExecutedAtLastCompleted(ShardingContexts shardingContexts);
  
  /** * Notify waiting task start. */
  public void notifyWaitingTaskStart() {
        
      synchronized (startedWait) {
        
          startedWait.notifyAll();
      }
  }
  
  /** * Notify waiting task complete. */
  public void notifyWaitingTaskComplete() {
        
      synchronized (completedWait) {
        
          completedWait.notifyAll();
      }
  }
  

}
```
从代码中可以看出,主要是重写了beforeJobExecuted和afterJobExecuted,The implementation principle of the two methods is the same,我们单从beforeJobExecuted分析执行步骤

• zk节点注册,The registered directory is /${job_name}/guarantee/started/${节点},Registered nodes are permanent nodes
• Determine whether all nodes have registered functions：zk节点/${job_name}/guarantee/started/Whether the number of child nodes is equal to the number of shards
• if【相等】Start executing the preceding method,and clear the node path,此时StartedNodeRemovedJobListener监听到节点变化,Wake up all object instances
• if【不相等】By using object locks,调用waitThe method first determines whether the object has timed out,If not timed out, the node path will be cleared,此时StartedNodeRemovedJobListener监听到节点变化,Wake up all object instances;并抛出异常;
  -若未超时,This node sleeps,and release the object lock,Other nodes acquire object locks,Re-judg whether all nodes have been registered successfully
  Summarize the above flow chart as：

总结

The listener is temporarily explained here,If there are problems we can continue to discuss,欢迎指教！