又被夺命连环问了!从一道关于定时任务的面试题说起。

发布时间 2023-12-18 13:02:12作者: why技术

你好呀,我是歪歪。

定时任务,大家在开发的过程中肯定都是接触过的。

歪师傅面试的时候关于定时任务一般都会问这样的一个问题:在实际开发的过程中,你们是如何避免定时任务重复执行的呢?

什么意思呢?

我给你上个图你就明白了。

假设我们有个订单服务的微服务,它部署在两台机器上:

这是一个再正常不过的部署方案了吧。

现在有一个需求来了:要从数据库里面获取前一日状态为成功的订单,然后把这些订单一笔笔的调用其他服务的接口,通知给他们。

写代码的时候很简单,基于 Quartz 框架,咔嚓几下就能搞出一个定时任务来,伪代码如下:

//每天10点触发一次
@Scheduled(cron = "0 0 10 * * ?")
public void sendOrder() {
    //查询前一日状态为成功的订单
    List<Order> orderList = selectSuccessOrder();
    for (Order order : orderList) {
        //发送订单到数据分析服务
        sendOrder(order);
    }
}

测试的时候也非常的正常,看不出任何毛病。

但是一上生产就完犊子了,为什么呢?

因为测试环境一般来说就只部署一台服务器,但是生产环境是多台呀:

每天 10 点一到,两台机器都跑起来了...

同样的逻辑跑了两次,一下就瓜起了涩,这肯定不是我们想要的结果。

问:这个情况你怎么处理?

在实际开发的过程中,我理解大家理论上都会遇到这个问题的。

歪师傅当年还是一个小萌新,第一次遇到这个问题的时候,是怎么考虑的呢?

抠了抠脑袋,想到一个自己觉得非常靠谱的解决方案。

各个微服务提供接口,接口内部实现定时任务的业务逻辑。然后抽离出一个专门的定时任务微服务,在这个服务中开发定时任务,来调用对应的接口:

由于“定时任务微服务”只部署在一台服务器上,所以当定时任务的时间一到,只会发起一次 RPC 调用,具体调用哪一台服务,由 RPC 的负载均衡来决定。

从而规避了前面提到的“触发两次”的问题。

当时我还觉得微服务的思想还是真是厉害,这样一抽离之后,业务代码和定时逻辑彻底分离开来,横向扩展也不需要考虑“多次触发”的问题:

但是,问题随着就来了:定时任务服务只部署了一台,它有单点风险啊,它挂了,所有的定时任务不就都挂了吗?

我知道在有的公司,实际情况就是这样的,知道服务有单点风险,但是评估下来,觉得是可以接受的,大不了就是做好服务监控,出了问题就赶紧重启一波服务。

所以遇到这个问题的解决方案就是:不管它。

但是,朋友,面试的时候你能这样回答吗,你是去调侃面试官的吗?

所以,该怎么办?

单点问题,很好解决,针对“定时任务服务”多部署一台服务器就行了:

但是,调用关系怎么办呢?

时间一到,咔的一下,两台“定时任务服务”都跑起来了,都对下游发起了 RPC 调用,这不又出现了前面这样“调用两次”的问题吗:

开始套娃了,你说怎么办?

这个时候歪师傅又抠了抠脑袋,又想到一个自己觉得非常靠谱的解决方案。

关于这个问题,我先换个壳问你:如果有多个请求过来,但是我们同一时间只想让一个请求正常执行,请问你怎么办?

一般来说我们都会想到加锁嘛。

单机的话,什么 synchronized,ReentrantLock 这些玩意就使劲儿往上怼。

多台服务就上分布式锁嘛,Redis、Zookeeper 拿出来秀一秀嘛,对不对?

比如,如果我们用 Redis,怎么做?

在发起 RPC 调用之前先从 Redis 里面拿锁,多台机器,谁拿到了,谁就可以执行:

//每天10点触发一次
@Scheduled(cron = "0 0 10 * * ?")
public void sendOrder() {
    //获取redis锁
    if(SET key value expireTime nx){
        //拿到锁的才能调用订单服务发送成功订单逻辑的接口
        callOrderRPC();   
    }
}

这样即使某一台服务器上的服务挂了,另外一台也能确保定时任务按时触发,并表示非常开心:很好,没人和我抢锁了。

或者说基于 zookeeper 来做。

比如我们定时任务的服务启动的时候,以服务名称维度向 zk 申请一个临时节点。

谁申请成功了,谁就算加锁成功了,虽然到点之后每个定时任务都会按时触发,但是和 Redis 同理,只有拿到锁的实例才能执行定时任务。

没有拿到锁的怎么办呢?

监听这个临时节点,处于随时待命状态。如果当前持有锁的服务挂了,那么临时节点也就没了,相当于锁就释放了,就可以上手抢锁了。

抢到锁,就可以执行定时任务,这样也能保证高可用。

如果是面试,针对“避免定时任务重复执行”能回答到分布式锁这里,我认为就可以了。

但是,朋友,这可是面试,面试一般是出连招的。

如果我突然画风一转,顺势提出下一个问题:

用分布式锁,可以通过只让一台机器运行的方式解决重复运行的问题。现在我换个场景,问问题,如果我昨日成功的订单数据量比较多,假设有 100w 笔吧,如果只在一台机器上跑,即使开启多线程,也需要很长的时间,而且是一台机器忙的不行,不太机器在旁边闲的不行。如果我想要充分把机器利用起来,让两台机器都来处理这 100w 笔订单,各自处理 50w 条,时间不就缩短了吗?

就像是这样:

请问,阁下又该如何应对?

ElasticJob

好了,前面铺垫了这么多,终于要引出 ElasticJob 这个玩意了。

这是官方文档的地址:

https://shardingsphere.apache.org/elasticjob/current/cn/overview/

其中有一个章节叫做“弹性调度”:

弹性调度是 ElasticJob 最重要的功能,也是这款产品名称的由来。 它是一款能够让任务通过分片进行水平扩展的任务处理系统。

从关于“分片”的描述中,我们知道也许能在这里找到问题的答案。

虽然答案就在眼前,但是别猴急。按照歪师傅的风格,还是得先上个 Demo 作为引子,给你抽丝剥个茧。

这里顺便吐槽一句官方文档:

你这个“快速入门”写的是什么玩意,根本就不能用好吧?

quick start 不能拿来即用,对于本白嫖党来说,是很难受的,好吗。

害得我还得自己摸索一下,还好整体并不复杂,你按照歪师傅给你提供的“快速入门”,五分钟足够搭个 Demo 了。

首先,新建一个 Spring Boot 项目,在 pom 文件中加入相关引用:

<dependency>
    <groupId>org.apache.shardingsphere.elasticjob</groupId>
    <artifactId>elasticjob-lite-spring-boot-starter</artifactId>
    <version>3.0.1</version>
</dependency>

然后实现 SimpleJob 接口,自定义一个定时任务:

package com.example.elasticjobtest;

@Slf4j
@Component
public class SpringBootJob implements SimpleJob {

    @Override
    public void execute(ShardingContext shardingContext) {
        log.info("SpringBootJob作业,分片总数是【{}】,当前分片是【{}】,分片参数是【{}】",
                shardingContext.getShardingTotalCount(),
                shardingContext.getShardingItem(),
                shardingContext.getShardingParameter());
    }
}

接着在 application.yml 里面添加配置:

elasticjob:
  # 注册中心配置
  regcenter:
    serverlists: 127.0.0.1:2181
    # ZooKeeper 的命名空间
    namespace: why-elastic-job
  # 作业配置
  jobs:
    springJob: # job的名称
      elasticJobClass: com.example.elasticjobtest.SpringBootJob
      cron: 0/5 * * * * ?
      shardingTotalCount: 2
      shardingItemParameters: 0=Beijing,1=Shanghai

就这几行代码,Demo 就算搭完了。

你自己说,这整个流程是不是五分钟够够的了?

在把服务启动起来之前,针对 application.yml 的配置,我先多 BB 几句。

里面这两个玩意是什么东西呢:

可以参考官方文档中的描述:

https://shardingsphere.apache.org/elasticjob/current/cn/user-manual/configuration/

shardingTotalCount 叫做作业分片总数,这个概念非常重要,理解了这个概念,就理解了 ElasticJob 的核心理念,先按下不表。

shardingItemParameters 叫做个性化分片参数,我这里写的是 0=Beijing,1=Shanghai,看起来很奇怪对不对,怎么突然冒出了北京和上海呢?

因为这也是官方文档中的案例:

这只是实例而已,当你理解了这个概念的用途之后,就可以按照自己的需求进行“个性化”配置。

Demo 跑起来

这个是 ElasticJob 的架构示意图:

可以看到它选择了 Zookeeper 做为自己的注册中心,所以在启动 Demo 之前,需要你把你本地的 Zookeeper 启动起来。

然后把 Demo 运行起来,观察日志输出:

你会发现每隔 5s 就会输出这样的日志:

2023-12-16 16:31:45.020 SpringBootJob作业,分片总数是【2】,当前分片是【0】,分片参数是【Beijing】
2023-12-16 16:31:45.020 SpringBootJob作业,分片总数是【2】,当前分片是【1】,分片参数是【Shanghai】

怎么样,看到日志输出之后是不是稍微品出了点淡淡的味道,就是那种虽然不知道怎么回事,但是总感觉马上就摸到门道的感觉。

保持住这种感觉,歪师傅马上就让你摸到门把手了。

为了模拟多个服务部署的情况,所以我们需要再多启动一个服务。

在 Idea 里面点击这个:

然后把“Allow multiple instances(运行多实例运行)”勾选上:

修改一下服务端口,避免端口冲突:

接着再次启动 Demo,观察一下日志:

标号为 ① 的地方是仅一台服务器运行的情况,两个分片都在这一个服务器上运行。

标号为 ② 和 ③ 的地方是两台服务器都运行起来的情况,同样的代码、同样的配置,跑在不同的端口而已。

一台的日志输出是这样的:

SpringBootJob作业,分片总数是【2】,当前分片是【1】,分片参数是【Shanghai】
SpringBootJob作业,分片总数是【2】,当前分片是【1】,分片参数是【Shanghai】

另外一台的日志输出是这样的:

SpringBootJob作业,分片总数是【2】,当前分片是【0】,分片参数是【Beijing】
SpringBootJob作业,分片总数是【2】,当前分片是【0】,分片参数是【Beijing】

可以看到,每隔五秒钟两台服务器都同时触发了定时任务,但是一台拿到的参数是 Shanghai,一台拿到的参数是 Beijing。

这个时候我们再回去看面试官的这个问题:

假设有 100w 笔吧,如果只在一台机器上跑,即使开启多线程,也需要很长的时间,而且是一台机器忙的不行,不太机器在旁边闲的不行。如果我想要充分把机器利用起来,让两台机器都来处理这 100w 笔订单,各自处理 50w 条,时间不就缩短了吗?

然后我再给你上个图:

每个机器上运行的代码是一样的,但是通过 ElasticJob 能让每个机器在运行定时任务的时候,拿到不一样的参数。

基于这个不一样的参数,我们就能搞很多事情了嘛。

比如 100w 数据,分为两组,一组 50w 条。假设 ID 是连续自增的,是不是可以这样判断奇偶数:

偶数:id % 2 == 0
奇数:id % 2 == 1

在这个表达式里面,每个数据的 id 是确定的,而这个“2”,你看它像不像是我们的“分片数”?至于这个“0”和“1”,是不是可以通过我们的“个性化分片参数”传递进来?

id % 分片数 == 个性化分片参数

比如我们写个这样的代码:

然后把作业配置改成这样的:

然后启动两个服务,我们观察一下日志输出:

一台机器处理的是 “1,3,5,7,9”,一台机器处理的是“0,2,4,6,8”

刚刚面试官的问题是啥来着?

两台机器处理 100w 笔订单,各自处理 50w 条?

这不就实现了吗?

再给你看一个神奇的东西,假设我在运行时把 shardingTotalCount 修改为 3,即分片数变成 3,对应的自定义参数也进行对应的修改,会发生什么事情呢?

按照我们之前的这个逻辑:

id % 分片数 == 个性化分片参数

0 到 9 这十个数字分别对 3 取模,那么就会分成下面这三组:

  • 第一组:0,3,6,9
  • 第二组:1,4,7
  • 第三组:2,5,8

这个没有任何毛病,对不对?

然后还需要特别注意的是,我说的是“在运行时”修改。

怎么修改?

很简单,ElasticJob 其实提供了对应的管理后台页面可以进行参数修改,但是我这里偷个懒,难得去部署对应的管理后台,,准备换个简单的思路。

因为前面说了,ElasticJob 使用的是 zk 做为自己的注册中心,我直接用工具连接上 zk,然后修改 zk 节点就行了。

我是怎么知道修改 zk 的哪个节点的呢?

别着急,等下就讲,歪师傅先带你看效果。

我这里用的工具是 ZooInspector,修改之后直接点击保存:

然后,朋友们,注意了,看日志输出

为了让你看的更加清楚,我把关键日志单独拿出来:

第一台机器上的日志是这样的:

分片总数是【3】,当前分片是【1】,分片参数是【1】,处理的数据 date=【1,4,7,】

第二台机器上的日志是这样的:

分片总数是【3】,当前分片是【0】,分片参数是【0】,处理的数据 date=【0,3,6,9,】
分片总数是【3】,当前分片是【2】,分片参数是【2】,处理的数据 date=【2,5,8,】

和我们前面推理的结果一模一样。

好,到这里就可以解答我的一个“按下不表”了。

首先,shardingTotalCount 叫做作业分片总数,在我前面的例子中,作业分片总数一共是 3 片:

  • 第一组(第一片):0,3,6,9
  • 第二组(第二片):1,4,7
  • 第三组(第三片):2,5,8

分成三片之后,Elasticjob 怎么知道每一片应该处理哪些数据呢?

它不知道,它也不用知道。它只需要告诉每一台服务器:“来,哥们,给你一个号你拿着。你们这波一共有多少多少个人,你是第几片。”

就完事了。

因为“昨日成功的订单”这个总的要处理的数据是不变的,所有每一台服务器知道一共要把这批数据分成几片,自己是第几片后,通过代码就能拿到对应的该处理的数据。

然后你再去看官方描述中关于“分片项”你大概就能知道这到底是个啥玩意了:

有的哥们比较猛,一次拿到两个号,也没关系,就是多处理一份数据嘛。这种情况就适用于两台机器的性能不一致的情况。

但是我用这个案例并不是为了引出“性能不一致”这种极少数的情况,而是为了这个...

当我再启动一个新的服务器,当第三台服务器加入之后,我们啥也没干,它自己就开始处理任务了。

3 个分片,一台服务器处理一个分片的数据。

能自动加入,就能自动退出,所以假设我把一台服务给关闭了:

从日志可以看出来,数据并没有丢。

第一台机器把本来该在下线的这台服务器上处理的数据给接管了:

分片总数是【3】,当前分片是【2】,分片参数是【2】,处理的数据 date=【2,5,8,】
分片总数是【3】,当前分片是【0】,分片参数是【0】,处理的数据 date=【0,3,6,9,】

好了,到这里,基本功能就算演示完成,可以适当的响起一些掌声了。

啥原理啊?

其实关于原理,官方文档上也按照步骤进行了比较详细的说明:

https://shardingsphere.apache.org/elasticjob/current/cn/features/elastic/

如果你不了解 zk 的大致工作原理、节点特性、监听机制啥的,后面肯定会看得比较懵逼。

所以需要先去补一下这方面的信息,对于这部分的描述和源码的解读有很大帮助。

如果你能大致理解 zk 的工作原理,那么整体读下来其实没有什么特别难以理解的地方,如果要深入理解每一个步骤的话,那肯定要读一下源码的。

步骤都有了,去找对应的源码,不就是按图索骥,手拿把掐的事情吗。

在阅读源码之前,还有一个非常重要的东西要铺垫一下,前面也说了:基于 zk 做的注册中心。

所以你必须要了解“注册中心的数据结构”是怎么样的,每个节点是干啥的,才能理解代码里面操作 zk 节点的时候,到底是什么含义。

关于注册中心的数据结构,文档上也有介绍:

我觉得这个还是非常重要的,所以我多啰嗦几句,主要给你看看实际的数据是怎么样的。

还是以我本地启动三个服务为例。

启动起来之后,看 zk 上注册了这些节点:

其中“why-elastic-job”和“springJob”分别是我们写在 application.yml 里面的 ZooKeeper 的命名空间和 Job 名称:

config 节点

config 节点里面是作业配置信息,以 YAML 格式存储:

可以看到节点里面实际的值比我们配置的多,因为有很多默认项。每个默认项是干啥的,就自己去研究吧。

前面我说的“运行时修改”,就修改的是这个地方信息。

我为什么知道改这里?

还不是官网告诉我的。

instances 节点

该节点是作业运行实例信息,子节点是当前作业运行实例的主键。

作业运行实例主键由作业运行服务器的 IP 地址和 PID 构成。

作业运行实例主键均为临时节点,当作业实例上线时注册,下线时自动清理。注册中心可以监控这些节点的变化,来协调分布式作业的分片以及高可用。

具体到我们这个案例中,是这样的:

instances 下面有三个子节点,代表有三个微服务。

假设我停止运行一个服务,由于是 zk 的临时节点,这个地方就会变成 2 个:

sharding 节点

作业分片信息,子节点是分片项序号,从零开始,至分片总数减一。比如我们这里就是 0 到 2:

分片项序号的子节点存储详细信息,每个分片项下的子节点用于控制和记录分片运行状态:

  • sharding-0-instance:192.168.2.16@-@4964
  • sharding-1-instance:192.168.2.16@-@2224
  • sharding-2-instance:192.168.2.16@-@4964

可以看到 0,2 分片是运行在同一个 instance 上的,这一点和日志是匹配的:

sharding 下除了 instance 节点外,可能还有其他的节点,详细信息说明如下:

servers 节点

作业服务器信息,子节点是作业服务器的 IP 地址。

可在 IP 地址节点写入 DISABLED 表示该服务器禁用。

在新的云原生架构下,servers 节点大幅弱化,仅包含控制服务器是否可以禁用这一功能。

为了更加纯粹的实现作业核心,servers 功能未来可能删除,控制服务器是否禁用的能力应该下放至自动化部署系统。

leader 节点

作业服务器主节点信息,下面有三个子节点:

  • election:用于主节点选举
  • sharding:用于分片
  • failover:用于失效转移处理

除了节点介绍外,在官网描述上有这样的一句话:

换句话说就是,如果你想了解作业,那这个节点是很重要的。看源码的时候,需要特别关注对于 leader 节点下的操作。

在我们的案例中,instance 里面的信息是这样的:

表示这个节点是主节点。

源码

知道了 zk 上每个节点的用处,看源码的时候比着看就行了。

源码比较多,歪师傅这里只能带着你做个非常简单的导读。

首先,因为很多逻辑都是基于 zk 节点在来做的,所以最重要的是各种各样的 zk 节点监听器,ElasticJob 在启动时,会执行这个方法,开启监听器:

org.apache.shardingsphere.elasticjob.kernel.internal.listener.ListenerManager#startAllListeners

比如前面说的这个节点:

如果这个节点存在,则说明需要重新分片,对应的监听器是这个:

shardingListenerManager.start();

那么什么时候会触发“重新分片”呢?

  • 如果分片总数变化,或作业服务器节点上下线或启用/禁用,以及主节点选举,会触发设置重分片标记
  • 作业在下次执行时使用主节点重新分片,且中间不会被打断作业执行时不会触发分片

所以在 shardingListenerManager 监听器里面我们可以看到这两个逻辑:

满足条件之后,就会执行设置重新分片标识的代码:

shardingService.setReshardingFlag();

该方法里面,创建了一个新的节点:

这个节点,就是它:

再比如,看看这个方法:

org.apache.shardingsphere.elasticjob.lite.internal.sharding.ShardingService#shardingIfNecessary

这个方法是做对作业进行分片逻辑的。

对作业进行分片,首先我们要知道当前有哪些实例在运行,对不对?

那怎么才能知道呢?

instances 节点请求出战:

shardingIfNecessary 方法的第一行逻辑就是读取 instances 节点下的数据:

获取到节点之后,是不是就可以分片了?

理论上是这样的,但是别着急,你看源码里面还有这样一个判断:

isLeaderUntilBlock,看方法名称也知道了,看看 Leader 节点是不是到位了,如果没到位,需要等一下 Leader 选举结束。

怎么判断 Leader 节点是不是到位了?

前面文档中说了,就是看这个节点是否存在:

对应到源码就是这样的:

所以这就是我前面说的,你看源码的时候得结合 zk 节点的用途一起看,知道节点的用途就能理解源码里面操作节点的目的是什么。

然后,在这里多说一句。

shardingIfNecessary 这个方法是读取配置,处理分片逻辑的。

但是这个方法在每一个实例中都会运行,岂不是每个实例都会执行一次分片逻辑?

这样处理的话,由于多个地方执行分片逻辑,就需要考虑冲突和一致性的问题,导致逻辑非常的复杂。

虽然这个方法每个实例都会执行,但是其实只需要有一个实例执行分片逻辑就行了。

那么哪个节点来执行呢?

你肯定也猜到了,当然是主节点来干这个事儿嘛。如果当前节点不是主节点 return 就完事了:

怎么看当前节点是否是主节点呢?

前面已经出现多次了,zk 里面记录着的:

如果当然节点是主节点,就接着往下执行,就是“作业分片策略”了:

目前官方提供了三个不同的分片策略:

对应的实现类是这样的:

逻辑都非常简单,上手 Debug 两次就能摸清楚。

建议直接把项目拉下来,然后从测试用例入手。

好了,源码导读就到这里了。

我觉得我已经算是告诉你关于 ElasticJob 源码阅读的方式和注意点,如果你掌握到了,留言区留言“清晰”二字,支持一波。

如果你还是云里雾里的,没事,是我的问题。大胆的说出来:什么玩意?看求不懂。呸,垃圾作者。

如果你是第一次接触到 ElasticJob,那么读到这里的时候,你的内心关于 ElasticJob 应该还有很多疑问以及不清楚的细节。

很好,带着你的问题,去翻源码吧。

源码之下无秘密。

下面这个环节叫做[荒腔走板],技术文章后面我偶尔会记录、分享点生活相关的事情,和技术毫无关系。我知道看起来很突兀,但是我喜欢,因为这是一个普通博主的生活气息。

荒腔走板

这周终于是把《长安三万里》看了,之前一直想看,但是又被三个小时的时长劝退。

我个人觉得确实是值得豆瓣高分的。

看完之后,包括看的过程中,我老是想起之前在网上看到的一段话,关于“一颗子弹”和“教育闭环”的。

“一颗子弹”是指在《我与地坛》看到的一段书评,其内容是:一个人十三四岁的夏天,在路上捡到一支真枪,因为年少无知,天不怕地不怕,他扣下扳机。没有人死,也没有人受伤,他认为自己开了空枪。后来他三十岁或者更老,走在路上听到背后隐隐约约的风声。他停下来回过身去,子弹正中眉心。

“教育闭环”是指教育具有长期性和滞后性,起初你只能理解表层的道理,直到多年后的某个瞬间,你才能真正领悟到书上知识的真谛,此时教育的任务才算真正完成。

我小时候读到“两岸猿声啼不住,轻舟已过万重山”的时候,重点总是在“两岸猿声”上,想象着猿猴的叫声是什么样的,那是一番怎样有趣的画面。

后来,甚至可以说是今年,这个电影上映之后,我才明白当年读书的时候我忽略的“轻舟已过万重山”背后才是有更加蜿蜒曲折、激动人心的故事。

这句诗就是当年的那一颗子弹,命中了马上三十岁的我,至此,教育才算完成了闭环。

今年,让我产生同样感受的,还有当年完全忽略的这句话:孔乙己是站着喝酒而穿长衫的唯一的人穿的虽然是长衫,可是又脏又破似平十多年没有补,也没有洗。他对人说话,总是满口之乎者也叫人半懂不懂的。

此外,电影中多次提到“长安”,虽然我们学的是同样的课本,读的是一样的诗,但是每个人对与“长安”的认知和理解是不一样的。

现在提到长安,我脑海中出现的第一个画面永远是当年看《河西走廊》纪录片的时候那一个画面。

第一集《使者》,张骞出使西域,被匈奴囚禁九年后同随从堂邑父出逃,继续西行。

靠强大意志力穿越塔克拉玛干沙漠和帕米尔高原,到达西域。回程再次被俘,数年后带匈奴妻子和堂邑父又一次出逃东归。

十三年后,终于再次望到长安城,张骞匍匐在地,长跪不起。

西北望长安,可怜无数山。

这一跪,看的我眼泪婆娑。