ElasticJob分片机制
Posted
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了ElasticJob分片机制相关的知识,希望对你有一定的参考价值。
参考技术A ElasticJob是一个弹性的分布式任务调度框架,这里的分布式就是采用分片的来进行任务调度和业务执行的解耦,分片信息就是中间进行解耦的。ElasticJob任务调度框架调度触发执行的是分片,然后业务可以在框架触发对应分片信息的时候,增加自己业务的处理。分片这个思想挺不错的,把任务调度框架和实际业务解耦的相当好。ShardingListenerManager分片管理监听器在ElasticJob启动的就是开启了监听,这里是开启了2个监听器ShardingTotalCountChangedJobListener(分片节点总数变化监听器)和ListenServersChangedJobListener(服务器改变监听器)。
获取zookeeper下发的分片个数变化事件的通知,判断新分片数和原分片数是否相等,不相等的话设置需要重新分片的标记,创建/leader/sharding/neccessary持久节点。
instance和servers节点下有子节点变化会被监听到,这个时候也会去设置下需要重新分片的标记/leader/sharding/neccessary节点。
上面是触发生成了需要重新分片的标记,具体分片的执行时在任务执行的过程中。在作业任务执行的时候需要获取分片信息,这个时候会完成重新分片的执行。
源码分析ElasticJob任务错过机制(misfire)与幂等性
任务在调度执行中,由于某种原因未执行完毕,下一次调度任务触发后,在同一个Job实例中,会出现两个线程处理同一个分片上的数据,这样就会造成两个线程可能处理相同的数据,因此Elastic-Job引入幂等机制来解决上述问题。再重申一次ElastciJob的分布式是数据的分布式,一个任务在多个Job实例上运行,每个Job实例处理该Job的部分数据(数据分片)。
本文重点分析ElasticJob是如何做到如下两点的。
ElasticJob如何确保在同一个Job实例中多个线程不会处理相同的数据
ElasticJob如何确保数据不会被多个Job实例处理
为了解决上述这种情况,ElasticJob引入任务错过补偿执行(misfire)与幂等机制。
ElasticJob幂等原理
场景:例如任务调度周期为每5s执行一次,正常每次调度任务处理需要耗时2s,如果在某一段时间由于数据库压力变大,导致原本只需要2s就能处理完成的任务,现在需要16s才能运行,在一批数据处理未完成的情况下,每5s又会触发一次调度,如果不加以控制的话,在同一个实例上根据分片条件去查询数据库,查询到的数据有可能相同(部分相同),这样同一条任务数据将被多次处理,如果业务方法未实现幂等,则会引发非常严重的问题,那ElasticJob是否可以避免这个问题呢?
答案是肯定。elasticJob提供了一个配置参数:monitorExecution=true,开启幂等性。
一个任务触发后,将执行任务处理逻辑,其入口:
1AbstractElasticJobExecutor#misfireIfRunning
2if (jobFacade.misfireIfRunning(shardingContexts.getShardingItemParameters().keySet())) { // @1
3 if (shardingContexts.isAllowSendJobEvent()) { // @2
4 jobFacade.postJobStatusTraceEvent(shardingContexts.getTaskId(), State.TASK_FINISHED, String.format(
5 "Previous job '%s' - shardingItems '%s' is still running, misfired job will start after previous job completed.", jobName,
6 shardingContexts.getShardingItemParameters().keySet()));
7 }
8 return;
9}
代码@1:在一个调度任务触发后如果上一次任务还未执行,则需要设置该分片状态为mirefire,表示错失了一次任务执行。
代码@2:如果该分片被设置为mirefire并开启了事件跟踪,将事件跟踪保存在数据库中。
接下来详细分析JobFacade.misfireIfRu-nning的实现逻辑:
1/**
2 * 如果当前分片项仍在运行则设置任务被错过执行的标记.
3 *
4 * @param items 需要设置错过执行的任务分片项
5 * @return 是否错过本次执行
6 */
7 public boolean misfireIfHasRunningItems(final Collection<Integer> items) {
8 if (!hasRunningItems(items)) {
9 return false;
10 }
11 setMisfire(items);
12 return true;
13 }
如果存在未完成的分片,则调用setMis-fire(items)方法,在开启monitorExecut-ion(true)的情况下,在分片任务开始时会创建{namespace}/jobname/sharding/{item}/running节点,在任务结束后会删除该目录,所以在判断是否有分片正在运行时,只需判断是否存在上述节点即可。如果存在,调用setMisfire方法。
PS:ElasticJob只有在monitorExecuti-on=true的情况下,才会创建{namespa-ce}/jobname/sharding/{item}/running,m-isfire机制才能生效。
1ExecutionService#setMisfire
2/**
3 * 设置任务被错过执行的标记.
4 *
5 * @param items 需要设置错过执行的任务分片项
6 */
7 public void setMisfire(final Collection<Integer> items) {
8 for (int each : items) {
9 jobNodeStorage.createJobNodeIfNeeded(ShardingNode.getMisfireNode(each));
10 }
11 }
其实现方式为分配给该实例下的所有分片创建持久节点{namespace}/jobname/shading/{item}/misfire节点,注意,只要分配给该实例的任何一分片未执行完毕,则在该实例下的所有分片都增加m-isfire节点,然后忽略本次任务触发,等待任务结束后再执行。
1AbstractElasticJobExecutor#execute
2execute(shardingContexts, JobExecutionEvent.ExecutionSource.NORMAL_TRIGGER);
3 while (jobFacade.isExecuteMisfired(shardingContexts.getShardingItemParameters().keySet())) {
4 jobFacade.clearMisfire(shardingContexts.getShardingItemParameters().keySet());
5 execute(shardingContexts, JobExecutionEvent.ExecutionSource.MISFIRE);
6}
在任务执行完成后检查是否存在{name-space}/jobname/sharding/{item}/misfire节点,如果存在,则首先清除misfie相关的文件,然后执行任务。
幂等实现方案总结:
在下一个调度周期到达之后,只要发现这个分片的任何一个分片正在执行,则为该实例分片的所有分片都设置为mis-fire,等任务执行完毕后,再统一执行下一次任务调度。
ElasticJob数据分片
ElasticJob基于数据分片,不同分片根据分片参数(人为配置),从数据库中查询各自数据(任务数据分片),如果当节点宕机,数据会重新分片,如果任务未执行完成,然后执行分片动作,数据是否会被不同的任务同时处理呢?
答案是不会,因为当节点宕机后是否需要重新分片事件监听器会监听到Job实例代表的节点删除,设置重新分片,在任务被调度执行具体处理逻辑之前,需要重新分片,重新分片的前提又是要所有的分片的任务全部执行完毕,这也依赖是否开启幂等控制(monitorExecution)。
如果开启,ElasticJob能感知正在执行处理的分片,重新分片需要等待当前所有任务全部运行完毕后才会触发,故不会存在不同节点处理相同数据的问题。
问答:
1、如果一个任务JOB的调度频率为每10s一次,在某个时间,该job执行耗时用了33s(平时只需执行5s),按照正常调度,应该后续会触发3次调度,那该job后执行完,会连续执行3次调度吗?
答案:在33s这次任务执行完成后,如果后面的任务执行在10s内执行完毕的话,只会触发一次,不会补偿3次,因为Ela-sticJob记录任务错失执行,只是创建了misfire节点,并不会记录错失的次数。
以上是关于ElasticJob分片机制的主要内容,如果未能解决你的问题,请参考以下文章