scalaAkka并发编程定时任务简易通信框架

　　1。Akka并发编程框架简介1。1Akka概述
　　Akka是一个用于构建高并发、分布式和可扩展的基于事件驱动的应用工具包。Akka是使用scala开发的库，同时可以使用scala和Java语言来开发基于Akka的应用程序。1。2Akka特性提供基于异步非阻塞、高性能的事件驱动编程模型内置容错机制，允许Actor在出错时进行恢复或者重置操作超级轻量级的事件处理（每GB堆内存几百万Actor）使用Akka可以在单机上构建高并发程序，也可以在网络中构建分布式程序。1。3Akka通信过程
　　以下图片说明了AkkaActor的并发编程模型的基本流程：学生创建一个ActorSystem通过ActorSystem来创建一个ActorRef（老师的引用），并将消息发送给ActorRefActorRef将消息发送给MessageDispatcher（消息分发器）MessageDispatcher将消息按照顺序保存到目标Actor的MailBox中MessageDispatcher将MailBox放到一个线程中MailBox按照顺序取出消息，最终将它递给TeacherActor接受的方法中2。创建Actor
　　Akka中，也是基于Actor来进行编程的。类似于之前学习过的Actor。但是Akka的Actor的编写、创建方法和之前有一些不一样。2。1API介绍ActorSystem：它负责创建和监督Actor1。在Akka中，ActorSystem是一个重量级的结构，它需要分配多个线程。2。在实际应用中，ActorSystem通常是一个单例对象，可以使用它创建很多Actor。3。直接使用context。system就可以获取到管理该Actor的ActorSystem的引用实现Actor类1。定义类或者单例对象继承Actor（注意：要导入akka。actor包下的Actor）2。实现receive方法，receive方法中直接处理消息即可，不需要添加loop和react方法调用。Akka会自动调用receive来接收消息。3。【可选】还可以实现preStart（）方法，该方法在Actor对象构建后执行，在Actor生命周期中仅执行一次。加载Actor1。要创建Akka的Actor，必须要先获取创建一个ActorSystem。需要给ActorSystem指定一个名称，并可以去加载一些配置项（后面会使用到）2。调用ActorSystem。actorOf（Props（Actor对象），Actor名字）来加载Actor。2。2ActorPath
　　每一个Actor都有一个Path，这个路径可以被外部引用。路径的格式如下：
　　Actor类型
　　路径
　　示例
　　本地Actor
　　akka：actorSystem名称userActor名称
　　akka：SimpleAkkaDemousersenderActor
　　远程Actor
　　akka。tcp：mysysip地址：portuserActor名称
　　akka。tcp：192。168。10。17：5678userserviceb2。3入门案例2。3。1需求
　　基于Akka创建两个Actor，Actor之间可以互相发送消息。2。3。2实现步骤创建Maven模块创建并加载Actor发送接收消息2。3。3创建Maven模块
　　使用Akka需要导入Akka库，这里我们使用Maven来管理项目，具体步骤如下：创建Maven模块。打开pom。xml文件，导入akkaMaven依赖和插件。dependenciesdependencygroupIdorg。scalalanggroupIdscalalibraryartifactIdversion｛scala。version｝versiondependencydependencygroupIdcom。typesafe。akkagroupIdakkaactor2。11artifactIdversion2。3。14versiondependencydependencygroupIdcom。typesafe。akkagroupIdakkaremote2。11artifactIdversion2。3。14versiondependencydependencygroupIdcom。itheimagroupIdsparkdemocommonartifactIdversion1。0SNAPSHOTversiondependencydependenciesbuildsourceDirectorysrcmainscalasourceDirectorytestSourceDirectorysrctestscalatestSourceDirectorypluginsplugingroupIdnet。alchim31。mavengroupIdscalamavenpluginartifactIdversion3。2。2versionexecutionsexecutiongoalsgoalcompilegoalgoaltestCompilegoalgoalsconfigurationdependencyfilearg｛project。build。directory｝。scaladependenciesargargsconfigurationexecutionexecutionspluginplugingroupIdorg。apache。maven。pluginsgroupIdmavenshadepluginartifactIdversion2。4。3versionexecutionsexecutionphasepackagephasegoalsgoalshadegoalgoalsconfigurationfiltersfilter：artifactexcludesexcludeMETAINF。SFexcludeexcludeMETAINF。DSAexcludeexcludeMETAINF。RSAexcludeexcludesfilterfilterstransformerstransformerimplementationorg。apache。maven。plugins。shade。resource。AppendingTransformerresourcereference。confresourcetransformertransformerimplementationorg。apache。maven。plugins。shade。resource。ManifestResourceTransformermainClassmainClasstransformertransformersconfigurationexecutionexecutionspluginpluginsbuild2。3。4创建并加载Actor
　　到这，我们已经把Maven项目创建起来了，后续我们都会采用Maven来管理我们的项目。接下来，我们来实现：
　　创建并加载Actor，这里，我们要创建两个Actor：SenderActor：用来发送消息ReceiverActor：用来接收，回复消息
　　具体步骤在srcmainscala文件夹下创建包：com。itheima。akka。demo在该包下创建两个Actor（注意：用object修饰的单例对象）。SenderActor：表示发送消息的Actor对象。ReceiverActor：表示接收消息的Actor对象。在该包下创建单例对象Entrance，并封装main方法，表示整个程序的入口。把程序启动起来，如果不报错，说明代码是没有问题的。
　　参考代码objectSenderActorextendsActor｛细节：在Actor并发编程模型中，需要实现act方法，想要持续接收消息，可通过loopreact实现。在Akka编程模型中，需要实现receive方法，直接在receive方法中编写偏函数处理消息即可。重写receive（）方法overridedefreceive：Receive｛casexprintln（x）｝｝objectReceiverActorextendsActor｛重写receive（）方法overridedefreceive：Receive｛casexprintln（x）｝｝objectEntrance｛defmain（args：Array〔String〕）｛1。实现一个ActorTrait，其实就是创建两个Actor对象（上述步骤已经实现）。2。创建ActorSystem两个参数的意思分别是：ActorSystem的名字，加载配置文件（此处先不设置）valactorSystemActorSystem（actorSystem，ConfigFactory。load（））3。加载ActoractorOf方法的两个参数意思是：1。具体的Actor对象。2。该Actor对象的名字valsenderActoractorSystem。actorOf（Props（SenderActor），senderActor）valreceiverActoractorSystem。actorOf（Props（ReceiverActor），receiverActor）｝｝2。3。5发送接收消息
　　思路分析使用样例类封装消息SubmitTaskMessage提交任务消息SuccessSubmitTaskMessage任务提交成功消息使用！发送异步无返回消息。
　　参考代码MessagePackage。scala文件中的代码记录发送消息的样例类。parammsg具体的要发送的信息。caseclassSubmitTaskMessage（msg：String）记录回执信息的样例类。parammsg具体的回执信息。caseclassSuccessSubmitTaskMessage（msg：String）Entrance。scala文件中的代码程序主入口。objectEntrance｛defmain（args：Array〔String〕）：Unit｛1。创建ActorSystem，用来管理所有用户自定义的Actor。valactorSystemActorSystem（actorSystem，ConfigFactory。load（））2。通过ActorSystem，来管理我们自定义的Actor（SenderActor，ReceiverActor）valsenderActoractorSystem。actorOf（Props（SenderActor），senderActor）valreceiverActoractorSystem。actorOf（Props（ReceiverActor），receiverActor）3。由ActorSystem给SenderActor发送一句话start。senderActor！start｝｝SenderActor。scala文件中的代码objectSenderActorextendsActor｛overridedefreceive：Receive｛1。接收Entrance发送过来的：startcasestart｛2。打印接收到的数据。println（SenderActor接收到：Entrance发送过来的start信息。）3。获取ReceiverActor的具体路径。参数：要获取的Actor的具体路径。格式：akka：actorSystem的名字user要获取的Actor的名字。valreceiverActorcontext。actorSelection（akka：actorSystemuserreceiverActor）4。给ReceiverActor发送消息：采用样例类SubmitTaskMessagereceiverActor！SubmitTaskMessage（我是SenderActor，我在给你发消息！。。。）｝5。接收ReceiverActor发送过来的回执信息。caseSuccessSubmitTaskMessage（msg）println（sSenderActor接收到回执信息：｛msg｝）｝｝ReceiverActor。scala文件中的代码objectReceiverActorextendsActor｛overridedefreceive：Receive｛1。接收SenderActor发送过来的消息。caseSubmitTaskMessage（msg）｛2。打印接收到的消息。println（sReceiverActor接收到：｛msg｝）3。给出回执信息。sender！SuccessSubmitTaskMessage（接收任务成功！。我是ReceiverActor）｝｝｝
　　输出结果SenderActor接收到：Entrance发送过来的start信息。ReceiverActor接收到：我是SenderActor，我在给你发消息！。。。SenderActor接收到回执信息：接收任务成功！。我是ReceiverActor3。Akka定时任务
　　需求：如果我们想要使用Akka框架定时的执行一些任务，该如何处理呢？
　　答：在Akka中，提供了一个scheduler对象来实现定时调度功能。使用ActorSystem。scheduler。schedule（）方法，就可以启动一个定时任务。3。1schedule（）方法的格式方式一：采用发送消息的形式实现。defschedule（initialDelay：FiniteDuration，延迟多久后启动定时任务interval：FiniteDuration，每隔多久执行一次receiver：ActorRef，给哪个Actor发送消息message：Any）要发送的消息（implicitexecutor：ExecutionContext）隐式参数：需要手动导入方式二：采用自定义方式实现。defschedule（initialDelay：FiniteDuration，延迟多久后启动定时任务interval：FiniteDuration每隔多久执行一次）（f：Unit）定期要执行的函数，可以将逻辑写在这里（implicitexecutor：ExecutionContext）隐式参数：需要手动导入注意：不管使用上述的哪种方式实现定时器，都需要导入隐式转换和隐式参数，具体如下：
　　导入隐式转换，用来支持定时器。
　　importactorSystem。dispatcher
　　导入隐式参数，用来给定时器设置默认参数。
　　importscala。concurrent。duration。3。2案例
　　需求定义一个ReceiverActor，用来循环接收消息，并打印接收到的内容。创建一个ActorSystem，用来管理所有用户自定义的Actor。关联ActorSystem和ReceiverActor。导入隐式转换和隐式参数。通过定时器，定时（间隔1秒）给ReceiverActor发送一句话。方式一：采用发送消息的形式实现。方式二：采用自定义方式实现。
　　参考代码案例：演示Akka中的定时器。objectMainActor｛1。定义一个Actor，用来循环接收消息，并打印。objectReceiverActorextendsActor｛overridedefreceive：Receive｛casexprintln（x）不管接收到的是什么，都打印。｝｝defmain（args：Array〔String〕）：Unit｛2。创建一个ActorSystem，用来管理所有用户自定义的Actor。valactorSystemActorSystem（actorSystem，ConfigFactory。load（））3。关联ActorSystem和ReceiverActor。valreceiverActoractorSystem。actorOf（Props（ReceiverActor），receiverActor）4。导入隐式转换和隐式参数。导入隐式转换，用来支持定时器。importactorSystem。dispatcher导入隐式参数，用来给定时器设置默认参数。importscala。concurrent。duration。5。通过定时器，定时（间隔1秒）给ReceiverActor发送一句话。方式一：通过定时器的第一种方式实现，传入四个参数。actorSystem。scheduler。schedule（3。seconds，2。seconds，receiverActor，你好，我是种哥，我有种子你买吗？。。。）方式二：通过定时器的第二种方式实现，传入两个时间，和一个函数。actorSystem。scheduler。schedule（0seconds，2seconds）（receiverActor！新上的种子哟，你没见过！嘿嘿嘿。。。）实际开发写法actorSystem。scheduler。schedule（0seconds，2seconds）｛receiverActor！新上的种子哟，你没见过！嘿嘿嘿。。。｝｝｝4。实现两个进程之间的通信4。1案例介绍
　　基于Akka实现在两个进程间发送、接收消息。WorkerActor启动后去连接MasterActor，并发送消息给MasterActor。MasterActor接收到消息后，再回复消息给WorkerActor。
　　4。2Worker实现
　　步骤创建一个Maven模块，导入依赖和配置文件。创建Maven模块创建启动WorkerActor。在srcmainscala文件夹下创建包，在该包下创建WorkerActor（单例对象的形式创建）。在该包下创建Entrance单例对象，里边定义main方法发送setup消息给WorkerActor，WorkerActor接收打印消息。启动测试。
　　参考代码WorkerActor。scala文件中的代码1。创建WorkActor，用来接收和发送消息。objectWorkerActorextendsActor｛overridedefreceive：Receive｛2。接收消息。casexprintln（x）｝｝Entrance。scala文件中的代码程序入口。当前ActorSystem对象的路径akka。tcp：actorSystem127。0。0。1：9999objectEntrance｛defmain（args：Array〔String〕）：Unit｛1。创建ActorSystem。valactorSystemActorSystem（actorSystem，ConfigFactory。load（））2。通过ActorSystem，加载自定义的WorkActor。valworkerActoractorSystem。actorOf（Props（WorkerActor），workerActor）3。给WorkActor发送一句话。workerActor！setup｝｝启动测试：右键，执行，如果打印结果出现setup，说明程序执行没有问题。4。3Master实现
　　步骤创建一个Maven模块，导入依赖和配置文件。创建Maven模块。创建启动MasterActor。在srcmainscala文件夹下创建包，在该包下创建MasterActor（单例对象的形式创建）。在该包下创建Entrance单例对象，里边定义main方法WorkerActor发送connect消息给MasterActorMasterActor回复success消息给WorkerActorWorkerActor接收并打印接收到的消息启动Master、Worker测试
　　参考代码MasterActor。scala文件中的代码MasterActor：用来接收WorkerActor发送的数据，并给其返回回执信息。负责管理MasterActor的ActorSystem的地址：akka。tcp：actorSystem127。0。0。1：8888objectMasterActorextendsActor｛overridedefreceive：Receive｛1。接收WorkerActor发送的数据caseconnect｛println（MasterActor接收到：connect！。。。）2。给WorkerActor回执一句话。sender！success｝｝｝Entrance。scala文件中的代码Master模块的主入口objectEntrance｛defmain（args：Array〔String〕）：Unit｛1。创建ActorSystem，用来管理用户所有的自定义Actor。valactorSystemActorSystem（actorSystem，ConfigFactory。load（））2。关联ActorSystem和MasterActor。valmasterActoractorSystem。actorOf（Props（MasterActor），masterActor）3。给masterActor发送一句话：测试数据，用来测试。masterActor！测试数据｝｝WorkerActor。scala文件中的代码（就修改了第3步）WorkerActor：用来接收ActorSystem发送的消息，并发送消息给MasterActor，然后接收MasterActor的回执信息。负责管理WorkerActor的ActorSystem的地址：akka。tcp：actorSystem127。0。0。1：9999objectWorkerActorextendsActor｛overridedefreceive：Receive｛1。接收Entrance发送过来的：setup。casesetup｛println（WorkerActor接收到：Entrance发送过来的指令setup！。）2。获取MasterActor的引用。valmasterActorcontext。system。actorSelection（akka。tcp：actorSystem127。0。0。1：8888usermasterActor）3。给MasterActor发送一句话。masterActor！connect｝4。接收MasterActor的回执信息。casesuccessprintln（WorkerActor接收到：success！）｝｝5。案例：简易版spark通信框架5。1案例介绍
　　模拟Spark的Master与Worker通信。一个Master管理多个Worker若干个Worker（Worker可以按需添加）向Master发送注册信息向Master定时发送心跳信息5。2实现思路构建Master、Worker阶段构建MasterActorSystem、Actor构建WorkerActorSystem、ActorWorker注册阶段Worker进程向Master注册（将自己的ID、CPU核数、内存大小（M）发送给Master）Worker定时发送心跳阶段Worker定期向Master发送心跳消息Master定时心跳检测阶段Master定期检查Worker心跳，将一些超时的Worker移除，并对Worker按照内存进行倒序排序多个Worker测试阶段启动多个Worker，查看是否能够注册成功，并停止某个Worker查看是否能够正确移除5。3工程搭建
　　需求
　　本项目使用Maven搭建工程。
　　步骤分别搭建以下几个项目，GroupID统一都为：com。yueda，具体工程名如下：
　　工程名
　　说明
　　sparkdemocommon
　　存放公共的消息、实体类
　　sparkdemomaster
　　AkkaMaster节点
　　sparkdemoworker
　　AkkaWorker节点导入依赖分别在三个项目下的srcmain，srctest下，创建scala目录。导入配置文件（资料包中的application。conf）修改Master的端口为7000修改Worker的端口为80005。4构建Master和Worker
　　需求
　　分别构建Master和Worker，并启动测试
　　步骤创建并加载MasterActor创建并加载WorkerActor测试是否能够启动成功
　　参考代码完成master模块中的代码，即：在srcmainscala下创建包：com。itheima。spark。master，包中代码如下：MasterActor。scala文件中的代码Master：用来管理多个Worker的。MasterActor的路径：akka。tcp：actorSystem127。0。0。1：7000objectMasterActorextendsActor｛overridedefreceive：Receive｛casexprintln（x）｝｝Master。scala文件中的代码程序入口：相当于我们以前写的MainActorobjectMaster｛defmain（args：Array〔String〕）：Unit｛1。创建ActorSystem。valactorSystemActorSystem（actorSystem，ConfigFactory。load（））2。通过ActorSystem，关联MasterActor。valmasterActoractorSystem。actorOf（Props（MasterActor），masterActor）3。启动程序，如果不报错，说明代码没有问题。｝｝
　　完成worker模块中的代码，即：在srcmainscala下创建包：com。itheima。spark。worker，包中代码如下：WorkerActor。scala文件中的代码WorkerActor的地址：akka。tcp：actorSystem127。0。0。1：7100objectWorkerActorextendsActor｛overridedefreceive：Receive｛casexprintln（x）｝｝
　　Worker。scala文件中的代码程序入口objectWorker｛defmain（args：Array〔String〕）：Unit｛1。创建ActorSystem。valactorSystemActorSystem（actorSystem，ConfigFactory。load（））2。通过ActorSystem，关联MasterActor。valworkerActoractorSystem。actorOf（Props（WorkerActor），workerActor）3。启动程序，如果不报错，说明代码没有问题。workerActor！hello｝｝5。5Worker注册阶段实现
　　需求
　　在Worker启动时，发送注册消息给Master。
　　思路分析Worker向Master发送注册消息（workerid、cpu核数、内存大小）随机生成CPU核（1、2、3、4、6、8）随机生成内存大小（512、1024、2048、4096）（单位M）Master保存Worker信息，并给Worker回复注册成功消息启动测试
　　具体步骤在sparkdemocommon项目的srcmainscala文件夹下创建包在WorkerActor单例对象中定义一些成员变量，分别表示：masterActorRef：表示MasterActor的引用。workerid：表示当前WorkerActor对象的id。cpu：表示当前WorkerActor对象的CPU核数。mem：表示当前WorkerActor对象的内存大小。cuplist：表示当前WorkerActor对象的CPU核心数的取值范围。memlist：表示当前WorkerActor对象的内存大小的取值范围。在WorkerActor的preStart（）方法中，封装注册信息，并发送给MasterActor。在MasterActor中接收WorkerActor提交的注册信息，并保存到双列集合中。。MasterActor给WorkerActor发送回执信息（注册成功信息。）。在WorkerActor中接收MasterActor回复的注册成功信息。
　　参考代码WorkerActor。scala文件中的代码WorkerActor的地址：akka。tcp：actorSystem127。0。0。1：7100objectWorkerActorextendsActor｛1定义成员变量，记录MasterActor的引用，以及WorkerActor提交的注册参数信息。privatevarmasterActorRef：ActorSelection表示MasterActor的引用。privatevarworkerid：String表示WorkerActor的idprivatevarcpu：Int表示WorkerActor的CPU核数privatevarmem：Int表示WorkerActor的内存大小。privatevalcpulistList（1，2，3，4，6，8）CPU核心数的取值范围privatevalmemlistList（512，1024，2048，4096）内存大小取值范围2。重写preStart（）方法，里边的内容：在Actor启动之前就会执行。overridedefpreStart（）：Unit｛3。获取Master的引用。masterActorRefcontext。actorSelection（akka。tcp：actorSystem127。0。0。1：7000usremasterActor）4。构建注册消息。workeridUUID。randomUUID（）。toString设置workerActor的idvalrnewRandom（）cpucpulist（r。nextInt（cpulist。length））memmemlist（r。nextInt（memlist。length））5。将WorkerActor的提交信息封装成WorkerRegisterMessage对象。varregisterMessageWorkerRegisterMessage（workerid，cpu，mem）6。发送消息给MasterActor。masterActorRef！registerMessage｝overridedefreceive：Receive｛casexprintln（x）｝｝MasterActor。scala文件中的代码Master：用来管理多个Worker的。MasterActor的路径：akka。tcp：actorSystem127。0。0。1：7000objectMasterActorextendsActor｛1。定义一个可变的Map集合，用来保存注册成功好的Worker信息。privatevalregWorkerMapcollection。mutable。Map〔String，WorkerInfo〕（）overridedefreceive：Receive｛caseWorkerRegisterMessage（workId，cpu，mem）｛2。打印接收到的注册信息println（sMasterActor：接收到worker注册信息，｛workId｝，｛cpu｝，｛mem｝）3。把注册成功后的保存信息保存到：workInfo中。regWorkerMapworkIdWorkerInfo（workId，cpu，mem）4。回复一个注册成功的消息。sender！RegisterSuccessMessage｝｝｝修改WorkerActor。scala文件中receive（）方法的代码overridedefreceive：Receive｛caseRegisterSuccessMessageprintln（WorkerActor：注册成功！）｝5。6Worker定时发送心跳阶段
　　需求
　　Worker接收到Master返回的注册成功信息后，定时给Master发送心跳消息。而Master收到Worker发送的心跳消息后，需要更新对应Worker的最后心跳时间。
　　思路分析编写工具类读取心跳发送时间间隔创建心跳消息Worker接收到注册成功后，定时发送心跳消息Master收到心跳消息，更新Worker最后心跳时间启动测试
　　具体步骤在worker的srcmainresources文件夹下的application。conf文件中添加一个配置。worker。heartbeat。interval5配置worker发送心跳的周期（单位是s）在worker项目的com。itheima。spark。work包下创建一个新的单例对象：ConfigUtils，用来读取配置文件信息。在WorkerActor的receive（）方法中，定时给MasterActor发送心跳信息。Master接收到心跳消息，更新Worker最后心跳时间。。
　　参考代码worker项目的ConfigUtils。scala文件中的代码objectConfigUtils｛1。获取配置信息对象。privatevalconfigConfigFactory。load（）2。获取worker心跳的具体周期valworker。heartbeat。intervalconfig。getInt（worker。heartbeat。interval）｝修改WorkerActor。scala文件的receive（）方法中的代码overridedefreceive：Receive｛caseRegisterSuccessMessage｛1。打印接收到的注册成功消息println（WorkerActor：接收到注册成功消息！）2。导入时间单位隐式转换和隐式参数importscala。concurrent。duration。importcontext。dispatcher3。定时给Master发送心跳消息。context。system。scheduler。schedule（0seconds，ConfigUtil。worker。heartbeat。intervalseconds）｛3。1采用自定义的消息的形式发送心跳信息。masterActorRef！WorkerHeartBeatMessage（workerId，cpu，mem）｝｝｝MasterActor。scala文件中的代码objectMasterActorextendsActor｛1。定义一个可变的Map集合，用来保存注册成功好的Worker信息。privatevalregWorkerMapcollection。mutable。Map〔String，WorkerInfo〕（）overridedefreceive：Receive｛接收注册信息。caseWorkerRegisterMessage（workId，cpu，mem）｛2。打印接收到的注册信息println（sMasterActor：接收到worker注册信息，｛workId｝，｛cpu｝，｛mem｝）3。把注册成功后的保存信息保存到：workInfo中。regWorkerMapworkIdWorkerInfo（workId，cpu，mem，newDate（）。getTime）4。回复一个注册成功的消息。sender！RegisterSuccessMessage｝接收心跳消息caseWorkerHeartBeatMessage（workId，cpu，mem）｛1。打印接收到的心跳消息。println（sMasterActor：接收到｛workId｝的心跳信息）2。更新指定Worker的最后一次心跳时间。regWorkerMapworkIdWorkerInfo（workId，cpu，mem，newDate（）。getTime）3。为了测试代码逻辑是否OK，我们可以打印下regWorkerMap的信息println（regWorkerMap）｝｝｝5。7Master定时心跳检测阶段
　　需求
　　如果某个worker超过一段时间没有发送心跳，Master需要将该worker从当前的Worker集合中移除。可以通过Akka的定时任务，来实现心跳超时检查。
　　思路分析编写工具类，读取检查心跳间隔时间间隔、超时时间定时检查心跳，过滤出来大于超时时间的Worker移除超时的Worker对现有Worker按照内存进行降序排序，打印可用Worker
　　具体步骤修改Master的application。conf配置文件，添加两个配置配置检查Worker心跳的时间周期（单位：秒）master。check。heartbeat。interval6master。check。heartbeat。timeout15在Master项目的com。itheima。spark。master包下创建：ConfigUtils工具类（单例对象），用来读取配置文件信息。在MasterActor中开始检查心跳（即：修改MasterActorpreStart中的代码。）。开启Master，然后开启Worker，进行测试。
　　参考代码Master项目的ConfigUtils。scala文件中的代码针对Master的工具类。objectConfigUtil｛1。获取到配置文件对象。privatevalconfig：ConfigConfigFactory。load（）2。获取检查Worker心跳的时间周期（单位：秒）valmaster。check。heartbeat。intervalconfig。getInt（master。check。heartbeat。interval）3。获取worker心跳超时的时间（秒）valmaster。check。heartbeat。timeoutconfig。getInt（master。check。heartbeat。timeout）｝MasterActor。scala文件的preStart（）方法中的代码5。定时检查worker的心跳信息overridedefpreStart（）：Unit｛5。1导入时间转换隐式类型和定时任务隐式变量importscala。concurrent。duration。importcontext。dispatcher5。2启动定时任务。context。system。scheduler。schedule（0seconds，ConfigUtil。master。check。heartbeat。intervalseconds）｛5。3过滤大于超时时间的Worker。valtimeOutWorkerMapregWorkerMap。filter｛keyval5。3。1获取最后一次心跳更新时间。vallastHeatBeatTimekeyval。2。lastHeartBeatTime5。3。2超时公式：当前系统时间最后一次心跳时间超时时间（配置文件信息1000）if（newDate（）。getTimelastHeatBeatTimeConfigUtil。master。check。heartbeat。timeout1000）trueelsefalse｝5。4移除超时的Workerif（！timeOutWorkerMap。isEmpty）｛如果要被移除的Worker集合不为空，则移除此timeOutWorkerMap注意：双列集合是根据键移除元素的，所以最后的。1是在获取键。regWorkerMaptimeOutWorkerMap。map（。1）｝5。5对worker按照内存大小进行降序排序，打印Worker。2获取所有的WorkInfo对象。valworkerListregWorkerMap。map（。2）。toList5。6按照内存进行降序排序。valsortedWorkerListworkerList。sortBy（。mem）。reverse5。7打印结果println（按照内存的大小降序排列的Worker列表：）println（sortedWorkerList）｝｝5。8多个Worker测试阶段
　　需求
　　修改配置文件，启动多个worker进行测试。
　　大白话：启动一个Worker，就修改一次Worker项目下的application。conf文件中记录的端口号，然后重新开启Worker即可。
　　步骤测试启动新的Worker是否能够注册成功停止Worker，测试是否能够从现有列表删除