分布式定时任务elastic-job(三)

2021-05-30 本文已影响0人后来丶_a24d

分布式定时任务系列

选举

主节点作用: 分配作业分片项，调解分布式作业不一致状态。因为elastic-job-lite是无中心化，通过jar提供的，所以需要主节点处理一些事情。这里不通过分布式锁处理这些问题，是为了避免每次操作都需要获取锁解锁，避免性能损耗。不同的作业是有可能有不同的主节点的。
失效转移使用zk分布式锁，是因为希望集群一起负载失效的分片。
选主时序图

选主时序图.png
这里用分布式锁防止并发，回调时判断hasLeader是因为一台服务选完主节点后释放锁，另外的服务拿到锁后还是会执行回调代码，所以要加这个判断

// LeaderService.java
// 选举主节点
public void electLeader() {
   // 通过回调的机制实现抽象公共代码部分
   jobNodeStorage.executeInLeader(LeaderNode.LATCH, new LeaderElectionExecutionCallback());
}

// JobNodeStorage.java
public void executeInLeader(final String latchNode, final LeaderExecutionCallback callback) {
   // try with resource, LeaderLatch实现closable, 完成后会自动调用close
   try (LeaderLatch latch = new LeaderLatch(getClient(), jobNodePath.getFullPath(latchNode))) {
       latch.start();
       latch.await();
       callback.execute();
   } catch (final Exception ex) {
       handleException(ex);
   }
}

// LeaderElectionExecutionCallback 回调
class LeaderElectionExecutionCallback implements LeaderExecutionCallback {
   @Override
   public void execute() {
       // 当前无主节点
       if (!hasLeader()) { 
           jobNodeStorage.fillEphemeralJobNode(LeaderNode.INSTANCE, JobRegistry.getInstance().getJobInstance(jobName).getJobInstanceId());
       }
   }
}

选取成功后会填充节点/${JOB_NAME}/leader/electron/instance

选主时机

初始化注册作业启动

public void registerStartUpInfo(final boolean enabled) {
   // 选举主节点
   leaderService.electLeader();
}

节点数据变化时, 主节点下线

class LeaderElectionJobListener extends AbstractJobListener {
   
   @Override
   protected void dataChanged(final String path, final Type eventType, final String data) {
       if (!JobRegistry.getInstance().isShutdown(jobName) && (isActiveElection(path, data) || isPassiveElection(path, eventType))) {
           leaderService.electLeader();
       }
   }
}

等待主节点选举完成

有些必须要主节点操作的任务，会等待主节点完成再继续

// LeaderService
public boolean isLeaderUntilBlock() {
   // 不存在主节点 && 有可用的服务器节点
   while (!hasLeader() && serverService.hasAvailableServers()) {
       log.info("Leader is electing, waiting for {} ms", 100);
       // 等待
       BlockUtils.waitingShortTime();
       if (!JobRegistry.getInstance().isShutdown(jobName)
               && serverService.isAvailableServer(JobRegistry.getInstance().getJobInstance(jobName).getIp())) { 
          // 当前服务器节点可用
           electLeader();
       }
   }
   // 返回当前节点是否是主节点
   return isLeader();
}

zk节点数据结构

推荐使用zooInspector连接zk查看数据

ZK数据节点.png
com.seeger.demo.service.MySimpleJob是job的名称，每个job都会有类似这种目录，但是目录底下子目录leader, servers, config, instances, sharding这种结构是一样的

config

config节点是存放配置的，就是存一些比如分片总数之类的总体配置

servers

各个服务实例ip

leader

包括一些分布式锁，分片之类的信息

leader.png

instances

实例节点

sharding

分片相关，一个任务分片后，不同分片可能存在不同服务上

sharding.png

分片

分片策略

JobShardingStrategy策略接口

public interface JobShardingStrategy {
    Map<JobInstance, List<Integer>> sharding(List<JobInstance> jobInstances, String jobName, int shardingTotalCount);
}

默认策略AverageAllocationJobShardingStrategy

基于平均分配算法的分片策略, 如果有3台作业节点

分成9片，则每台作业节点分到的分片是：1=[0,1,2], 2=[3,4,5], 3=[6,7,8]
分成8片，则每台作业节点分到的分片是：1=[0,1,6], 2=[2,3,7], 3=[4,5]

感兴趣可以深入源码看看

OdevitySortByNameJobShardingStrategy

根据哈希值奇偶数确定ip升降序算法，奇数升序，偶数降序

如果有3台作业节点, 分成2片, 作业名称的哈希值为奇数, 则每台作业节点分到的分片是: 1=[ ], 2=[1], 3=[0].
如果有3台作业节点, 分成2片, 作业名称的哈希值为偶数, 则每台作业节点分到的分片是: 1=[0], 2=[1], 3=[ ]

优点: 可以根据作业名称重新分配作业节点负载, 不同的作业平均分配负载至不同的作业节点

RotateServerByNameJobShardingStrategy

根据作业名哈希值进行轮转的分片策略

如果有3台作业节点，分成2片，作业名称的哈希值为奇数，则每台作业节点分到的分片是：1=[0], 2=[1], 3=[]
如果有3台作业节点，分成2片，作业名称的哈希值为偶数，则每台作业节点分到的分片是：3=[0], 2=[1], 1=[]

优点分片更均衡, 不同的作业平均分配负载至不同的作业节点，确定分片数小于作业作业节点数时, 作业将永远分配至IP地址靠前的作业节点, 导致IP地址靠后的作业节点空闲

自定义实现

当然也可以自定义实现，实现时，@Bean传入自定义实现的配置，原理跟listen自定义实现一样。

分片条件

设置重新分片代码

设置要重新分片标记，zk上面的节点数据是/${JOB_NAME}/leader/sharding/necessary

// ShardingService.java
// 设置需要重新分片的标记.
public void setReshardingFlag() {
   jobNodeStorage.createJobNodeIfNeeded(ShardingNode.NECESSARY);
}

// JobNodeStorage.java
// 如果存在则创建作业节点
public void createJobNodeIfNeeded(final String node) {
   if (isJobRootNodeExisted() && !isJobNodeExisted(node)) {
       regCenter.persist(jobNodePath.getFullPath(node), "");
   }
}

判断是否需要重新分片, 判断/${JOB_NAME}/leader/sharding/necessary节点存不存在

// ShardingService.java
// 判断是否需要重分片 判断/${JOB_NAME}/leader/sharding/necessary节点存不存在
public boolean isNeedSharding() {
   return jobNodeStorage.isJobNodeExisted(ShardingNode.NECESSARY);
}

需要重新分片的情况

作业启动时

// SchedulerFacade.java
public void registerStartUpInfo(final boolean enabled) {
   // 设置 需要重新分片的标记
   shardingService.setReshardingFlag();
}

分片总数变化时

在监听器章节，介绍了监听器相关，分片总数变化时会触发ShardingTotalCountChangedJobListener监听器

// ShardingTotalCountChangedJobListener
class ShardingTotalCountChangedJobListener extends AbstractJobListener {
    
    @Override
    protected void dataChanged(final String path, final Type eventType, final String data) {
        if (configNode.isConfigPath(path) && 0 != JobRegistry.getInstance().getCurrentShardingTotalCount(jobName)) {
            int newShardingTotalCount = LiteJobConfigurationGsonFactory.fromJson(data).getTypeConfig().getCoreConfig().getShardingTotalCount();
            if (newShardingTotalCount != JobRegistry.getInstance().getCurrentShardingTotalCount(jobName)) {
                // 分片总数变化时设置需要重新分片
                shardingService.setReshardingFlag();
                JobRegistry.getInstance().setCurrentShardingTotalCount(jobName, newShardingTotalCount);
            }
        }
    }
}

服务器变化，服务器变化时触发服务器变化监听器

// ShardingListenerManager.java
class ListenServersChangedJobListener extends AbstractJobListener {

   @Override
   protected void dataChanged(final String path, final Type eventType, final String data) {
       if (!JobRegistry.getInstance().isShutdown(jobName)
               && (isInstanceChange(eventType, path)
                   || isServerChange(path))) {
           // 设置需要重新分片
           shardingService.setReshardingFlag();
       }
   }
   
   private boolean isInstanceChange(final Type eventType, final String path) {
       return instanceNode.isInstancePath(path) && Type.NODE_UPDATED != eventType;
   }
   
   private boolean isServerChange(final String path) {
       return serverNode.isServerPath(path);
   }
}

自诊断修复时也会触发，具体在自诊断修复章节

作业分片

作业分片时序图

作业分片时序图.png

// ShardingService.java
// 如果需要分片且当前节点为主节点, 则作业分片 如果当前无可用节点则不分片
public void shardingIfNecessary() {
   // 获取可用job实例
   List<JobInstance> availableJobInstances = instanceService.getAvailableJobInstances();
   // 判断是否需要重新分片
   if (!isNeedSharding() || availableJobInstances.isEmpty()) {
       return;
   }
   // 非主节点等待作业分片项分配完成 主节点则继续
   if (!leaderService.isLeaderUntilBlock()) { 
       blockUntilShardingCompleted();
       return;
   }

   // 主节点继续，但是等待其他运行中的作业运行完
   waitingOtherJobCompleted();
   // 从zk获取配置
   LiteJobConfiguration liteJobConfig = configService.load(false);
   int shardingTotalCount = liteJobConfig.getTypeConfig().getCoreConfig().getShardingTotalCount();
   // 重新分片标记设置
   jobNodeStorage.fillEphemeralJobNode(ShardingNode.PROCESSING, "");
   // 重置作业分片项信息
   resetShardingInfo(shardingTotalCount);
   // zk事务执行多条语句
   JobShardingStrategy jobShardingStrategy = JobShardingStrategyFactory.getStrategy(liteJobConfig.getJobShardingStrategyClass());
   jobNodeStorage.executeInTransaction(new PersistShardingInfoTransactionExecutionCallback(jobShardingStrategy.sharding(availableJobInstances, jobName, shardingTotalCount)));
}

3. blockUntilShardingCompleted

非主节点等待作业分片项分配完成主节点则继续

4. waitingOtherJobCompleted

主节点继续，但是等待其他运行中的作业运行完，作业是否则运行中需要作业设置LiteJobConfiguration.monitorExecution = true才可以监控到

5. load

从zk获取配置而不走缓存，需要获取最新的数据

6. fillEphemeralJobNode

设置正在重新分片标记 /${JOB_NAME}/leader/sharding/processing

7. resetShardingInfo

重置分片信息

private void resetShardingInfo(final int shardingTotalCount) {
  // 重置 有效的作业分片项
  for (int i = 0; i < shardingTotalCount; i++) {
      // 移除 `/${JOB_NAME}/sharding/${ITEM_ID}/instance`
      jobNodeStorage.removeJobNodeIfExisted(ShardingNode.getInstanceNode(i)); 
      // 创建 `/${JOB_NAME}/sharding/${ITEM_ID}`
      jobNodeStorage.createJobNodeIfNeeded(ShardingNode.ROOT + "/" + i); 
  }
  // 移除 多余的作业分片项
  int actualShardingTotalCount = jobNodeStorage.getJobNodeChildrenKeys(ShardingNode.ROOT).size();
  if (actualShardingTotalCount > shardingTotalCount) {
      for (int i = shardingTotalCount; i < actualShardingTotalCount; i++) {
          // 移除 `/${JOB_NAME}/sharding/${ITEM_ID}`
          jobNodeStorage.removeJobNodeIfExisted(ShardingNode.ROOT + "/" + i); 
      }
  }
}

8. executeInTransaction zk提供的事务执行

JobShardingStrategy.sharding 分配分配策略前文介绍过了，

9. execute执行

设置 /JOB_NAME/sharding/ITEM_ID/instance 分片节点数据

// PersistShardingInfoTransactionExecutionCallback.java
class PersistShardingInfoTransactionExecutionCallback implements TransactionExecutionCallback {
   
   // 作业分片项分配结果缓存
   private final Map<JobInstance, List<Integer>> shardingResults;
   
   @Override
   public void execute(final CuratorTransactionFinal curatorTransactionFinal) throws Exception {
       // 设置 每个节点分配的作业分片项 真正设置数据进去
       for (Map.Entry<JobInstance, List<Integer>> entry : shardingResults.entrySet()) {
           for (int shardingItem : entry.getValue()) {
               curatorTransactionFinal.create().forPath(jobNodePath.getFullPath(ShardingNode.getInstanceNode(shardingItem))
                       , entry.getKey().getJobInstanceId().getBytes()).and();
           }
       }
       // 移除 作业需要重分片的标记、作业正在重分片的标记
       curatorTransactionFinal.delete().forPath(jobNodePath.getFullPath(ShardingNode.NECESSARY)).and();
       curatorTransactionFinal.delete().forPath(jobNodePath.getFullPath(ShardingNode.PROCESSING)).and();
   }
}

获取作业分片上下文集合

时序图

获取作业分片上下文集合.png
图中2, 6部分在失效转移章节介绍
具体代码

// LiteJobFacade.java
@Override
public ShardingContexts getShardingContexts() {
   // 失效转移部分，先跳过
   boolean isFailover = configService.load(true).isFailover();
   if (isFailover) {
       List<Integer> failoverShardingItems = failoverService.getLocalFailoverItems();
       if (!failoverShardingItems.isEmpty()) {
           return executionContextService.getJobShardingContext(failoverShardingItems);
       }
   }
   // 作业分片, 如果需要分片且当前节点为主节点
   shardingService.shardingIfNecessary();
   // 获取分配在本机的作业分片项
   List<Integer> shardingItems = shardingService.getLocalShardingItems();
   // 失效转移部分，先跳过
   if (isFailover) {
       shardingItems.removeAll(failoverService.getLocalTakeOffItems());
   }
   // 移除被禁用的作业分片项
   shardingItems.removeAll(executionService.getDisabledItems(shardingItems));
   // 获取当前作业服务器分片上下文
   return executionContextService.getJobShardingContext(shardingItems);
}

4. shardingIfNecessary

作业分片, 如果需要分片且当前节点为主节点则分片

5. getLocalShardingItems

获取分配在本机的作业分片项，即 /JOB_NAME/sharding/${ITEM_ID}/instance

// ShardingService.java
// 获取运行在本作业实例的分片项集合
public List<Integer> getLocalShardingItems() {
   if (JobRegistry.getInstance().isShutdown(jobName) || !serverService.isAvailableServer(JobRegistry.getInstance().getJobInstance(jobName).getIp())) {
       return Collections.emptyList();
   }
   return getShardingItems(JobRegistry.getInstance().getJobInstance(jobName).getJobInstanceId());
}

// 获取作业运行实例的分片项集合.
public List<Integer> getShardingItems(final String jobInstanceId) {
   JobInstance jobInstance = new JobInstance(jobInstanceId);
   if (!serverService.isAvailableServer(jobInstance.getIp())) {
       return Collections.emptyList();
   }
   List<Integer> result = new LinkedList<>();
   int shardingTotalCount = configService.load(true).getTypeConfig().getCoreConfig().getShardingTotalCount();
   for (int i = 0; i < shardingTotalCount; i++) {
       // `/${JOB_NAME}/sharding/${ITEM_ID}/instance` 如果是本机则添加
       if (jobInstance.getJobInstanceId().equals(jobNodeStorage.getJobNodeData(ShardingNode.getInstanceNode(i)))) {
           result.add(i);
       }
   }
   return result;
}

7. shardingItems.removeAll

移除被禁用的作业分片项

// ExecutionService.java
// 获取禁用的任务分片项
public List<Integer> getDisabledItems(final List<Integer> items) {
   List<Integer> result = new ArrayList<>(items.size());
   for (int each : items) {
       // /${JOB_NAME}/sharding/${ITEM_ID}/disabled
       if (jobNodeStorage.isJobNodeExisted(ShardingNode.getDisabledNode(each))) {
           result.add(each);
       }
   }
   return result;
}

getJobShardingContext获取分片上下文

移除正在运行中的作业分片项，monitor开启才会记录运行中分片，才能有效移除正在运行的分片

// ExecutionContextService.java
public ShardingContexts getJobShardingContext(final List<Integer> shardingItems) {
   LiteJobConfiguration liteJobConfig = configService.load(false);
   // 移除 正在运行中的作业分片项，monitor开启才会记录运行中分片
   removeRunningIfMonitorExecution(liteJobConfig.isMonitorExecution(), shardingItems);
   if (shardingItems.isEmpty()) {
       return new ShardingContexts(buildTaskId(liteJobConfig, shardingItems), liteJobConfig.getJobName(), liteJobConfig.getTypeConfig().getCoreConfig().getShardingTotalCount(), 
               liteJobConfig.getTypeConfig().getCoreConfig().getJobParameter(), Collections.<Integer, String>emptyMap());
   }
   // 解析分片参数
   Map<Integer, String> shardingItemParameterMap = new ShardingItemParameters(liteJobConfig.getTypeConfig().getCoreConfig().getShardingItemParameters()).getMap();
   // 创建 分片上下文集合
   return new ShardingContexts(buildTaskId(liteJobConfig, shardingItems), //
           liteJobConfig.getJobName(), liteJobConfig.getTypeConfig().getCoreConfig().getShardingTotalCount(),
           liteJobConfig.getTypeConfig().getCoreConfig().getJobParameter(),
           getAssignedShardingItemParameterMap(shardingItems, shardingItemParameterMap)); // 获得当前作业节点的分片参数
}

失效转移

失效转移，指的是，别的服务宕机，这时候本服务运行完时会主动拉取失效转移的，当然也有被动触发的
失效转移会调用quartz的trigger 会再次触发一次等执行完再次触发，多次触发确实是串行执行的，因为设置了org.quartz.threadPool.threadCount设置quartz线程数为1，可以在自定义任务里面sleep然后多线程调用trigger做测试，可以发现是串行执行
失效转移涉及别的服务宕机之类的，涉及重新分片更复杂，所以是在重新执行任务的时候做错过转移，而不像错过执行一样，是具备运行完马上再执行的

作业节点崩溃监听

根据上问的监听器介绍，作业节点崩溃是在JobCrashedJobListener做监听，如果开启失效转移并且/JOB_NAME/instances/INSTANCE_ID节点被移除，先获取/JOB_NAME/sharding/ITEM_ID/failover节点数据，如果没有的话获取/JOB_NAME/sharding/ITEM_ID/instance节点数据

// JobCrashedJobListener.java
class JobCrashedJobListener extends AbstractJobListener {
   
   @Override
   protected void dataChanged(final String path, final Type eventType, final String data) {
       // 开启失效转移 并且${JOB_NAME}/instances/${INSTANCE_ID}节点被移除
       if (isFailoverEnabled() && Type.NODE_REMOVED == eventType
               && instanceNode.isInstancePath(path)) { 
           String jobInstanceId = path.substring(instanceNode.getInstanceFullPath().length() + 1);
           if (jobInstanceId.equals(JobRegistry.getInstance().getJobInstance(jobName).getJobInstanceId())) {
               return;
           }
           // 先获取${JOB_NAME}/sharding/${ITEM_ID}/failover节点数据
           List<Integer> failoverItems = failoverService.getFailoverItems(jobInstanceId);
           if (!failoverItems.isEmpty()) {
               for (int each : failoverItems) {
                   failoverService.setCrashedFailoverFlag(each);
                   failoverService.failoverIfNecessary();
               }
           } else {
                // 没有的话获取/${JOB_NAME}/sharding/${ITEM_ID}/instance节点数据
               for (int each : shardingService.getShardingItems(jobInstanceId)) { 
                   failoverService.setCrashedFailoverFlag(each);
                   failoverService.failoverIfNecessary();
               }
           }
       }
   }
}

两个节点其中一个有数据时, 设置fail转移标识

public void setCrashedFailoverFlag(final int item) {
   if (!isFailoverAssigned(item)) {
       // /${JOB_NAME}/leader/failover/items/${ITEM_ID}
       jobNodeStorage.createJobNodeIfNeeded(FailoverNode.getItemsNode(item)); 
   }
}
    
private boolean isFailoverAssigned(final Integer item) {
   return jobNodeStorage.isJobNodeExisted(FailoverNode.getExecutionFailoverNode(item));
}

作业失效转移

如果需要进行失效转移，利用分布式锁，然后回调机制实现

// FailoverService.java
public void failoverIfNecessary() {
   if (needFailover()) {
       jobNodeStorage.executeInLeader(FailoverNode.LATCH, new FailoverLeaderExecutionCallback());
   }
}

private boolean needFailover() {
    // `${JOB_NAME}/leader/failover/items/${ITEM_ID}` 有失效转移的作业分片项而且当前作业不在运行中
    return jobNodeStorage.isJobNodeExisted(FailoverNode.ITEMS_ROOT) && !jobNodeStorage.getJobNodeChildrenKeys(FailoverNode.ITEMS_ROOT).isEmpty()
            && !JobRegistry.getInstance().isJobRunning(jobName);
}

FailoverLeaderExecutionCallback回调逻辑, 最后需要的话会调用jobScheduleController.triggerJob()重新触发任务，这里参数设置了串行执行，所以不会有并发问题，因为失效转移比错过执行更复杂，往往涉及重分片，所以失效转移是重新出发任务，而错过执行是执行任务完马上再执行，而不重新触发新任务执行

class FailoverLeaderExecutionCallback implements LeaderExecutionCallback {
   
   @Override
   public void execute() {
       // 判断需要失效转移
       if (JobRegistry.getInstance().isShutdown(jobName) || !needFailover()) {
           return;
       }
       // 获得一个 `${JOB_NAME}/leader/failover/items/${ITEM_ID}` 作业分片项
       int crashedItem = Integer.parseInt(jobNodeStorage.getJobNodeChildrenKeys(FailoverNode.ITEMS_ROOT).get(0));
       // 设置这个 `${JOB_NAME}/sharding/${ITEM_ID}/failover` 作业分片项 为 当前作业节点
       jobNodeStorage.fillEphemeralJobNode(FailoverNode.getExecutionFailoverNode(crashedItem), JobRegistry.getInstance().getJobInstance(jobName).getJobInstanceId());
       // 移除这个 `${JOB_NAME}/leader/failover/items/${ITEM_ID}` 作业分片项
       jobNodeStorage.removeJobNodeIfExisted(FailoverNode.getItemsNode(crashedItem));
       // 触发作业执行
       JobScheduleController jobScheduleController = JobRegistry.getInstance().getJobScheduleController(jobName);
       if (null != jobScheduleController) {
           jobScheduleController.triggerJob();
       }
   }
}

核心代码executor, 执行完任务时会主动进行failoverIfNecessary的方法，有需要则执行FailoverLeaderExecutionCallback回调，并设置失效转移/JOB_NAME/sharding/ITEM_ID/failover节点。所以getFailoverItems获取失效转移分片时是先获取/JOB_NAME/sharding/ITEM_ID/failover节点顺序，因为这个是执行完任务设置的

// AbstractElasticJobExecutor
public final void execute() {
   // 执行 普通触发的作业
   execute(shardingContexts, JobExecutionEvent.ExecutionSource.NORMAL_TRIGGER);
   // 执行 被跳过触发的作业
   while (jobFacade.isExecuteMisfired(shardingContexts.getShardingItemParameters().keySet())) {
       jobFacade.clearMisfire(shardingContexts.getShardingItemParameters().keySet());
       execute(shardingContexts, JobExecutionEvent.ExecutionSource.MISFIRE);
   }
   
   // 执行 作业失效转移
   jobFacade.failoverIfNecessary();
}

failoverIfNecessary只会抓取一个失效分片，这样集群就能一起处理失效分片，而不是所有任务都打到某台负载

获取作业分片上下文集合

时序图

获取作业分片上下文集合.png
失效转移时获取上下文，分片介绍了大部分流程，这里只介绍2，6, 2在作业节点崩溃监听章节介绍过了，执行完任务时有需要会设置失效转移/JOB_NAME/sharding/ITEM_ID/failover节点。所以getFailoverItems获取失效转移分片时是先获取/JOB_NAME/sharding/ITEM_ID/failover节点顺序。 6是移除失效转移分片

// LiteJobFacade.java
@Override
public ShardingContexts getShardingContexts() {
   // 获得 失效转移的作业分片项
   boolean isFailover = configService.load(true).isFailover();
   if (isFailover) {
       // 2. 获取失效转移分片
       List<Integer> failoverShardingItems = failoverService.getLocalFailoverItems();
       if (!failoverShardingItems.isEmpty()) {
           // 跳过因为在分片章节介绍过了
           return executionContextService.getJobShardingContext(failoverShardingItems);
       }
   }
   // 跳过因为在分片章节介绍过了
   shardingService.shardingIfNecessary();
   // 跳过因为在分片章节介绍过了
   List<Integer> shardingItems = shardingService.getLocalShardingItems();
   // 6. 移除失效转移分片
   if (isFailover) {
       shardingItems.removeAll(failoverService.getLocalTakeOffItems());
   }
   // 移除 被禁用的作业分片项
   shardingItems.removeAll(executionService.getDisabledItems(shardingItems));
   // 跳过因为在分片章节介绍过了
   return executionContextService.getJobShardingContext(shardingItems);
}

1. 移除失效转移分片，这么做是因为有可能: 作业节点A持有作业分片项(0, 1)，此时断网，导致(0, 1)被作业节点B失效转移抓取，此时若作业节点A恢复，作业分片项(0, 1)依然属于作业节点A，但是可能已经在作业节点B执行，因此需要进行移除，避免多节点运行相同的作业分片项

shardingItems.removeAll(failoverService.getLocalTakeOffItems());
// 获取运行在本作业服务器的被失效转移的序列号
public List<Integer> getLocalTakeOffItems() {
    List<Integer> shardingItems = shardingService.getLocalShardingItems();
    List<Integer> result = new ArrayList<>(shardingItems.size());
    for (int each : shardingItems) {
        if (jobNodeStorage.isJobNodeExisted(FailoverNode.getExecutionFailoverNode(each))) {
            result.add(each);
        }
    }
    return result;
}

错过执行

错过任务执行指的是: 任务时间间隔小于任务执行时间，比如时间间隔10秒，但是任务执行本身花了15秒，这个时候下个任务触发时，因为有任务在执行，就会被表过错过
错过执行本身quartz会支持，但是elastic-job设置quartz处理逻辑交还到ealstic-job本身，错过执行是在执行完正常的任务之后，立马执行错过任务，而不是重新执行任务时，执行错过执行任务。executor方法开始时防御性判是否有错过任务，应该只是防御性的判断这块。当然错过执行执行时也会判断任务是否在运行中
错过执行只是涉及自己机器而且是在运行中才会错过，具备运行完马上再执行的条件。但是失效转移涉及别的服务宕机之类的，涉及重新分片更复杂，所以是在重新执行任务的时候做错过转移，而不像错过执行一样，是具备运行完马上再执行的

分布式定时任务elastic-job(三)

目录

分布式定时任务系列

选举

选主时机

等待主节点选举完成

zk节点数据结构

config

servers

leader

instances

sharding

分片

分片策略

默认策略AverageAllocationJobShardingStrategy

OdevitySortByNameJobShardingStrategy

RotateServerByNameJobShardingStrategy

自定义实现

分片条件

设置重新分片代码

需要重新分片的情况

作业分片

3. blockUntilShardingCompleted

4. waitingOtherJobCompleted

5. load

6. fillEphemeralJobNode

7. resetShardingInfo

8. executeInTransaction zk提供的事务执行

9. execute执行

获取作业分片上下文集合

4. shardingIfNecessary

5. getLocalShardingItems

7. shardingItems.removeAll

getJobShardingContext获取分片上下文

失效转移

作业节点崩溃监听

作业失效转移

获取作业分片上下文集合

错过执行

参考文章

猜你喜欢

热点阅读