247-尚硅谷-Scala核心编程-Master检测心跳并删除超时的Worker.avi原创

呃，那么下面呢，我们要完成另外一个任务，就是什么呢？Master，我们需要master启，呃启动一个定时任务，定时检测注册的worker，什么意思呢？是这样子的，就是你看这个图有了，我们分析起来比较到位，大家都知道我们这个master这边呢，呃，维护了很多的这个，呃哈希map，维护了一个哈希map，哈希map里面呢，坦白讲就是这个哈希map里面，它有很多很多的这个work，就是那个worker信息，比如说如果我们画一个图来说的话，就相当于这样有一这样一种这样一种形式，我给他。啊，把这个打开一下。啊，比如说我们这边有这么一，诶这这个这个颜色还改回去啊，我新加一个啊，我新加一个，先把这个含义说清楚，比如说我这里呢，有一个图形。对，这个这个就是我们的哈希map。

啊，就是就是我们的那个，就是我们所说的workers。这个work呢，就是我们的所说的一个map，这里面呢，管理了很多这个worker，比如说worker ID，然后呢，这是它的一个信息。啊，这是它的一个信息。啊，然后呢，第这是一个，比如说这是一一号啊呃，第二这是第二个第二个。这是他第二个。啊，可能还有第三个，第四个我就不写那么多了，比如说我就写三个。啊，第三个，那现在有个问题，你的这个哈希map是跟他关联起来的。啊，你是跟他关联起来，那么假如这个客户端对应的这个worker，它已经很久许久没有更新，没有这个心跳了。没有心跳，你站在这个地方也没有意义，所以说呢，我们可以把它清空，因为你将来这个master管理的worker呢，肯定是有一些作用的，对吧？啊，那可以把它干掉，就是干掉，那我怎么样把它干掉呢？好，这就需要我们这边要处理一下什么呢？处理定时定时的检测。

检测，呃，就是管理的，哎，管理的这个worker的心跳。Walker心跳是否正常？心跳是否正常？对，那如果不正常，如果已经超时了。啊，就说他已经好久没有心跳了，你就把它怎么样呢，删除啊，就是说我们要做到这个任务啊，如果超时了，心跳超时。超时干什么呢，就删除。就删除，所以说所以说这个地方有一个。这个大家一看就知道，这里我肯定要去启动一个什么呢，这里我自己要去启动一个这样的东西，就是我要启动一个定时任务。让他在这里反复的帮我们做一件事情，就是去检测这个东西。

啊，也就是说这个地方我们会启动一个定时检测的一个任务。对，你得去分析定时检测任务干什么呢？去检测它有没有超时的。所以这边相当于说是一个定时的任务，要做这个事情了，我写到这一栏。啊，比如这是啊，我们认为是个定时任务。哦，我用这个小线画一下啊。好，我用一个颜色标注一下，就是这是我的定时任务。定时器。就是定时检测。定时好简单写到这好，现在呢，注意现在我们完成这个功能跟客户端没有半毛钱关系了。跟他没有任何关系。也就现在检测这个任务只是发生在。服务器这边的。服务器，比如说master这边，那问题来了，我们怎么去启动这个定时器呢？

好的，我认为是在哪里比较合适呢？我认为是在这启动比较合适，就是一旦我的这个master启动了，我就去检测。我就去检测，在这里触发一个。呃，触发一个这个这个消息，让他去干，知会他，或者是让我这个master定时的启动这个任务，好，现在我的思路分析就完毕了啊，那思路分析完毕以后呢，我们自然就开始走代码，找到我们的master。好，呃，同学们可以看到，因为你在这个要在这个地方触发，那也就是说实际上我的代码应该这样写。啊，这里。这里开始开始写啊，开始写。啊，那怎么办呢？简单的很，我们把这个定时任务这样处理一下，好大家看我这里呢，设计了一个。啊，设计了，呃，设计了一个这个协议，这个协议呢，我就拿来用一下啊，我这已经有这个协议了，我这里设计了一个协议啊，这样协议其实可以并做一步啊，我我先写出来，待会儿呢，要简化也可以简化啊。

我这里写了两个协议。啊，一个是master给自己发送一个触发检查超时worker信息，另外一个是。呃，每隔一定时间去呃，检测心跳，呃，如果说超时就删除，就remove timeout worker。这个地方呢，是按照Spark这个格式来写的，其实这地方有点啰嗦，其实可以直接定时触发这个请求就可以了啊，但是呢，Spark他有这个代码，就沿用了他的这个风格啊，其实有点多余，但是呢就用他的吧啊，Master给自己发一个出发，那现在呢就简单啊，两个都是object。那就他不用带信息，只是代表我要干什么事情。

好，那现在呢，我们就开始来写这段代码啊，写这代码好，呃，怎么写呢？大家看大体大体来说，大体来说就是在这里是大的地方，我先发一个这个信息过去，然后在这里出发一个请求，然后呢，这边就干相应的事情，好代码就很简单啊，那么开始写一下，找到我们的master，我在这儿先发一个消息。S。给发一个什么消息呢？发一个这个消息，就是刚才我写的start timeout这个消息。这个消息给谁了呢？啊，显然是给他自己改了。啊，给他自己了，给他自己的话呢，我为了跟他不破坏，我就把这个请求写到最后啊，写到最后case，如果我收到了start time out。这个这个信息我就我就知道人家准备要开始检测了啊，这样写呢也有好处，就是将来可能在这个阶段我们要做一些工作，所以说分开呢。

也应该是有它的道理的，那我就这样写，我就写说开始开始了这个呃，定时定时检测，检测我卡心跳的心跳的这个这个这个任务。好的。那你怎么去？定时启动了，那思路就跟原先是一样的，就是这段代码我们拿过来用一下啊，啊其实用这个应该就可以了，好把这个都拿，因为他这边用到了一个啊影视转换，所以说呢，我把这个拿过来啊，思路一样的，只是我这要改一改。我首先要引一个刚才需要的包，就是干了。是看到点current啊，Concurrent这个duration这个时间单位引进去，那这边呢，Heartbeat要改了，改成什么呢？诶改成我们的这个叫remove timeout worker把它写进去。

啊写进去，那这个地方我们不要三秒了，人家本身三秒更新一次，你也三秒检查一下呢，这个其实有点太快了。给他一点充裕时间，比如说我们我们九秒钟检测一次也是可以的啊，九秒因为不用那么频繁，对吧，九秒检测一次，大不了就说他他已经死了，可能有稍稍有个延时几秒钟，倒也不是大事啊，倒也不是大事，一个人就是好像一个人死了，这这个几秒钟埋葬无所谓啊，无所谓啊，这个九秒钟，我们就九秒钟这个去检测一次。好，那这个地方我们触发的是什么呢？触发的是这个，好，显然每这个地方只会被触发一次，这个定时器就启用了，那启用过后，下面呢，我们就来对这个removed timeout这个消息进行一个处理，消息处理。

好的，这个怎么处理呢？Very的easy，找到这个removed，因为它是一个object，所以说直接写就可以了。好，同学们看这面稍稍有一点麻烦，要动脑筋了哟。这个remove time out的任务是什么？先搞清楚，就是这里需要去检查，这里需要检查，检查哪些worker心跳超时了。心跳超时。超市，并从map中删除。这是我们要干的事儿。那这里面呢，我们要充分利用呃，SC的这个函数式编程的特点来完成它，这个应该很简单吧。很简单。好，首先我们先来确定一下，我们怎么算才是一个超时呢？哎，大家想一想，怎么算才是一个超时呢？是不是应该这样处理，我拿到现在的时间。

减去你的每一个，减去你的那个上一次更新时间，如果他们时间大于了一个值，我们认为就是超时了，对吧，理论上说，理论上说它只要它只要大于三秒钟就应该算是超时了，是不是因为你三秒更新一次嘛，但是呢，为了这个保证，嗯，有这个，呃，这个网络延时，因为你你虽然是三秒三秒跟人家蹦了一下，但是你这个心跳蹦到。蹦到这个遥远的地方，它可能有个时间嘛，你不能严格按三秒，所以说我们干脆设一个六秒，你三秒钟六秒我我觉得应该是足够的了，而且理论上我们这些服务器呢，都是在一一个一个机房，他也不会说哦，你这有个worker master到哪去，到美国去了，那倒也不至于好，这边我们思路就出来了啊啊心跳超时，比如六秒。呃，这个地方我们要用怎么处理呢？就是now，现在的时间减去它的last heartbeat。

Heartbeat这个time时间，Heart heartbeat heartbeat这个时间，如果它大于了6000。超时好代码开始写了。第一步注意听，首先。首先将所有的workers的这个信息取出来，是不是要把所有workers这个workers in for取出来呀，这个前面这个其实我们并不关心，主要是这个家伙是不是因为这个里面的它的ID和这个时寸是相同的，还记得吗？哎，所以说你要放心大胆的写，将这个里面的，说白了就是将这个work里面的。Works里面的这个所有的这个取出来，这个好取吗？非常的好取，大家知道我们在前面选map的时候有个values。这个Y60大家知道是取出了所有的这个内容是吧，我就写个叫worker。Worker for啊，就很多对吧，很多全部去，嗨，这个特别讨厌啊，这个地方we are。

VR好，我先这样自己重新写一遍啊。我一车就行了是吧，哎，他为什么老这样子。这个讨厌死了啊，Worker。Work work work啊，Worker in force拿到了，那拿到过后你要干一些什么事情呢？咱们还要获取到当前的时间，这个no没因为他就要用啊好，现在呢，我们获取一个now time用什么呢？就用我们的这一个，刚才用的这个叫做system。啊。啊，点这个current，他好，这个就是当前时间，我们六一个啊拿一个信息。诶，这个写到这来啊，叫v now time。现在这个时间是一个毫，是一个毫秒，看清楚了。那现在我们要怎么把它干掉呢？好，我们这样子边先过滤，是不是这样，我的思路是这样子啊，先把这样子啊，先把超时的，超时的所有。

这个worker。Inform渠道。就是先把所有超市的worker群取到，然后删除即可，删除即可，这怎么写呢？大家看我的代码啊，我这样写了，Worker就是所有的点the。Fair，那filter你看这里面人家已经告诉你了，这个filter里面要返回一个bird，首先出一个worker in for，这个work in for就是里面那个里，它这这不是知道的吗？对吧，就是遍历出来workers in for里面含的是worker in worker in for嘛。就是这边有很多混的work，所以我前面加了个S，这边就没有S了，那就应该怎么写呢，就是worker。For work，诶这个没有加S啊走。啊，干什么呢？就是你的这个我现在的time减去你这个worker in for里边的哪个呢？Last，呃，Last time，如果这两个这两个值啊，这两个的值相差这个值它大于了6000。

啊，说明什么，就是这个条件一旦满足的话，这个就是我要准备干掉的。这个worker in for要清楚的，这个work in for能理解吗？这段代码如果想不起来的话，想一想老师讲这个高阶函数传这个是怎么玩的啊，那想不起来自己再去琢磨琢磨，然后点for each for each，我要干什么呢？For each，我要把这个过滤出来的，因为它整个就返回一个集合了嘛，我才会把它干掉，那怎么干掉呢？我把这个working for拿掉就行了，大家看代码很简单。哦，怎么写呢？好，这边为了好写，我换，我先这样子啊，我先写完再换哈。啊，就这么写的，那我这边拿到一个，他这边又拿到一个worker in for。

啊，因为里面就是worker。Worker。啊，这边我要干件什么事情呢？走一段代码，走一段代码怎么呢，把它干掉。把这个人干掉，怎么干掉呢？是worker，是点remove啊remove remove呢，KK是从哪来的呢？就是从你这边取出来的ID，好，这样就写完了。好，这个代码呢，稍微的有一点长哈，有点长，我看这个地方能不能，呃，这个能不能换一行啊，这个点换一行啊。啊，这样子就可以了啊，这样看的也很清晰，大家再再再再说一遍啊，再说一遍这个代码可能看起来有点吃力啊，就说这个地方是我现在这个map里面所有的work for，你可理解成这个working for是哪个呢？你可以理解成就这一堆这个东西。就是。

能理解吗？就这一堆啊，没没有这个ID啊，没有这个ID，把这个取出来的目的是干什么呢？先把这个超时的这一堆东西先取出来，怎么取的呢？就是给它传入了一个匿名的函数。只要这个条件。满足，这个就是要被我干掉的。啊不，那么这个work for是哪来的，是从你一个work遍历出来啊，那个每个取出来的，好，然后这个filter完了过后，这里面这个这段代码执行过后呢，就是全部要准备删掉的。因为过滤嘛，过滤完了我就再对我们便利出来进行一个便利过后，这里面取出来的仍然是我可以，当然有些东说这个名字随便写，可不可以假设我叫这个，你这改成这个也是一样的。啊，代码要通，也就是说这个地方只是表示你你这个集合传进来的那一个值而已，只是我为了呃名称方便，我取的是work info。

甚至你前面这个不叫working for，这只要对应就行了，这点要理解啊，千万不要理解，就是诶，这个名字是固定的，也不要认为这个名字和这个名有关系。啊，没有从那个名字上没关系，只是他便利的那个对象呢，是啊，一个意思，好这边就拿到过，我去又传了一个逆面函数干什么呢，只要你传进来，这个我判断只要你这个满。只要是这个家伙，我就给你干掉完事了，代码写完好，这个地方我们打一句话，打一个什么呢？当前有多少个呢？有这么多个好有Dollar把它把它整出来啊，加上我们workers.size。啊，Size就是我们求这个map的一个大小啊，当前有这么多个，呃，Worker还是活存活的。好，这样看看有没有变化，待会work worker这个存活。

好，现在我们来跑一把，看看这个代码呢有没有问题，有没有问题，我们执行一下啊，看这个代码有没有毛病。好，这段代码你放心，写完了过后你也不用去检测你的客户端，因为目前这个检测任务跟你客户端没有任何关系，代码就写完了，来同学们我们跑一下，注意啊，我这设了个九九百，呃呃，这个九秒，这边设了六秒，是把这个时间稍微的宽裕了一点。考虑了他的一个网络延时。啊，考虑了它一个颜色这样子呢。安全一点好，同学们现在呢，我们跑一下来走一个代码。关闭，对，关闭，启动我们的master，启动我们master来玩一把。启动一下我们master。好，同学们请看，现在一旦启动过后呢，他就去检测了，说现在有零个worker存活对的吗？九秒钟更新一次吗？现在九秒过后，它又会弹一个，以当节有零个看对的。

啊，因为九秒我检测一下发现，呃，现在确实一个都没有，那现在呢，我让这个worker粉墨登场。来吧，啊，这这写错了啊，这个worker粉末中登场，他说来了，注册一把。注册一把，那这个家伙注册一把呢，好的。它启动了，它启动了啊，诶，他也他也不停的在发送心跳，这边应该检测到有一个worker来了，当前有一个worker是存活的，看到没有。因为我每隔三秒说这边会打出三个，然后对应一个，对应一次，这个输出好，马上输出，但就有一个，现在我让他死掉。假设因为种种原因，这个worker被被终止了，或者宕机了，断掉，断掉过后呢，一时半会他也检测不出来，好看一下代码。啊，它有一个hi worker存活，但下一个应该检测出来就没有了。好，再看零个存活，说明他已然检测到了这个这个这个信息没有没任何问题啊，没有任何问题，那当然有些同学老师我写两个可不可以也可以，你写两个已然可以解决这问题，好，这是我们关于这一个心跳检测任务的一个分析和完成，现在呢，我把刚才讲解的这个逻辑给大家简单的阐述一下。

诶，代码都比较简单啊，没有什么难度，没有什么难度，但是呢，虽然没什么难度，你自己去写，那你要也要动动脑筋的啊好的同学们，那刚才我们讲的是又一个功能，第三个功能把它搞定它。好的同学们来，刚才我们讲的是实现功能三对不对，好的朋友们马上将其搞定啊，来，标题三标题三我做了什么事情呢？我首先呃，做了一个我们要干什么事，做了一个功能说明，然后呢，做了一个功能说明。然后呢，做了一个思路的分析。啊，最后呢，我们做了一个代码的实现，诶就这么一个流流程，那么思路分析我是怎么分析的呢？我是这样分析的。

分析只看我们的服务端。准确的讲就是这一块。啊，我分析的就这块儿。这一块是我分析的。好的。然后呢，我把这一段分析出来的示意图呢，给同学们放到这里来啊，有点大，把它缩小一点啊，就是分析，那么代码实现呢，也异常的简单，只改了一个地方，就是服务器端啊不还加了一个加了加了协议啊，所以说呢，我们来看一下。好，第一步我们更改了，就是更新了，更新了什么呢？就是我们那个message。Message protocol protoc.scanner，对不对，那这段代码呢？诶，就是刚才老师在这写的代码，没问题，我拉过来就是加了这么一点东西。加了一个star这个，其实你也想想，这边其实完全不用这么麻烦是吧，直接就remove定时调用一个remove out也可以，但是呢，Spark它有这个中中间的一个，这个相当于做了一个过渡，也无所谓啊，不是什么大的问题，就保留了。

好的，这是它的这段代码，那么紧接着呢，我们还有一段代码，就是在我们的master这边，我们增加了这么一点东西啊，在这加了这样一句话，加了这句话，然后呢，下面增加了，呃，Start timeout启动了我们定时器这边和一个好把它也简单的处理一下。好的，这是这样一段代码。来给大家整一下啊，这边是更新了，更新了我们的Spark。Spark master.scan对不对，然后呢，这边一段代码再加了一点东西。哪里呢，就是这么一点啊，代码都非常的简单。好，这段代码应该是从哪截到这啊看一下。哦对，是从这儿写到这就可以了，没有少啊，没有少可以的同学们，这个呢，给大家放到笔记里面去。

就是这么两点，好，同学们，关于这个叫做定时检测超时worker并删除的任务，我们就说完了。