使用Scala的Akka通信编写spark底层通信框架、启动多个Worker、WebSocket的通信过程 28

缺乏、安全感 2022-01-26 08:34 236阅读 0赞

# 图解&简单介绍Spark底层通信框架 #

## 1.1 图解 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70]

*  角色分类：
    
     *  老大：Master
     *  小弟：Worker
 *  启动过程
    
     *  启动Master
     *  再启动Worker
     *  小弟找老大进行注册
     *  小弟要发送心跳，目的：告诉老大，我这个worker还活着
     *  老大定时的清除挂掉的超时的Worker。

## 1.2 结合代码分析 ##

*  a --worker 发送注册信息到 masterSS
    
     *  worker 发消息给 master, 需要调用context获取到master的引用对象, 使用引用对象发消息  
        master 发消息给 worker, 直接使用sender ! 就能发送
     *  hashMap相关操作
        
         *  hashMap.put(key) 存值
         *  hashMap(key) 得到的是value对应的类型的值  
            hashMap.get(key) 得到的是Option类型的值
 *  a – b-- master 接收到注册消息, 返回worker注册成功的消息
 *  c— worker接收到master反馈的注册成功的消息后,定时发送心跳给master, 报活
    
     *  定时任务
        
         *  java— Time TimeTask
         *  linux —crontab
         *  scala–Akka中, 需要先发送给自己中转一下 context.dispatcher

//定时单位需要导入此包
    import scala.concurrent.duration._
    //定时执行需要隐式转换 
    import context.dispatcher
    context.system.scheduler.schedule(0 millis,SEND_HEART_BEAT_INTERVAL millis,self,SendHeartBeat(workerId))

*  c-- master接收心跳, 存储到hashMap中, 同时也存到listBuffer中
 *  d–master定时检查worker是否还活着, 自己发给自己, 初始化方法中发送开启定时任务.同时对listBuffer中的worker按照memory降序排列

## 1.3 WebSocket通信过程(C/S) ##

*  通过客户端发送**http的请求**，来连接服务端
 *  服务端收到请求
 *  客户端和服务端简历了长连接，就可以直接通信。
 *  服务端的数据需要实时的推送到客户端。

# 2. 代码&启动多个Worker中IDEA配置 #

## 2.1 代码 ##

*  Master类

import akka.actor.{ Actor, ActorRef, ActorSystem, Props}
    import com.typesafe.config.{ Config, ConfigFactory}
    
    import scala.collection.mutable
    import scala.collection.mutable.ListBuffer
    
    //导入定时任务单位包
    import scala.concurrent.duration._
    
    /** * spark主节点 */
    
    //1. 继承actor
    class SparkMaster extends Actor { 
    
      //b. 先定义两个集合,一个用于存储worker发送过来的消息,一个用于后续对worker排序用消息
      //hashSet 存储worker发送过来的注册消息
      //listBuffer 存储检查心跳后存储的消息
      val hashMap = new mutable.HashMap[String, WorkerInfo]()
      val listBuffer = new ListBuffer[WorkerInfo]()
    
      //d. 定义定时时间, 要大于心跳时间,
      val CHECT_HEART_BEAT_INTERVAL = 15000
    
      //1.1 初始化方法
      override def preStart(): Unit = { 
          //d. 初始化定时器, 打开定时开关, 就打开一次就行
        import context.dispatcher
        context.system.scheduler.schedule(0 millis, CHECT_HEART_BEAT_INTERVAL millis, self, CheckOutTimer)
    
      }
    
      //1.2 接收消息的方法, 可以被调用多次
      override def receive: Receive = { 
        //a. 接收worker注册消息
        case SendRegistMes(workId, memory, cores) => { 
          println(s"接收到wokerId=${workId}的注册消息")
    
          //将消息存储到hashSet中, key就是workId, value就是workerInfo对象
          val workerInfo = new WorkerInfo(workId, memory, cores)
          //判断hashSet中是否存储了workId, 如果没有存储,就put()到map中
          if (hashMap.contains(workId) == false) { 
            hashMap.put(workId, workerInfo)
    
            //************还要经woker对象存到list中,用于排序使用
            listBuffer += workerInfo
    
            //b. master给worker反馈注册成功消息
            sender ! RegisteredMsg(s"workerId=$workId 注册成功")
    
          }
        }
    
        //c. 收到worker发送过来的心跳, 需要将接收到的心跳时的时间封装到hashMap中
        case HeartBeat(wokerId) => { 
          //判断wokerId是否存在于hashMap中
          if (hashMap.contains(wokerId)) { 
            //获取当前时间
            val currentTimeMillis: Long = System.currentTimeMillis()
    
            //从hashMap中获取到workerInfo对象
            val workerInfo: WorkerInfo = hashMap(wokerId)
    
            //将时间赋值给字段
            workerInfo.time_interval = currentTimeMillis
    
    // println(s"master--${currentTimeMillis}--接收到了workerId=${wokerId}的心跳")
    
          }
    
        }
    
        //d. 定时检查心跳并清除超时的worker信息
        case CheckOutTimer => { 
          //过滤掉超时的worker信息 当前时间-最近心跳时间 > 15s 就死了
          val OutTimeWorker: ListBuffer[WorkerInfo] = listBuffer.filter(x => System.currentTimeMillis() - x.time_interval > CHECT_HEART_BEAT_INTERVAL)
    
          //从hashMap&listBuffer中移除超时的worker
          for(worker <- OutTimeWorker) { 
            val workerId: String = worker.workerId
            hashMap.remove(workerId)
    
            listBuffer -= worker
    
            println(s"超时的worker为: $workerId")
          }
    
          //答应活着 的worker
          println("活着的worker个数: " + listBuffer.size)
        }
    
          //根据worker中的内存大小排序
        println(listBuffer.sortBy(x => x.memory).reverse)
    
    
      }
    }
    
    object SparkMaster { 
      def main(args: Array[String]): Unit = { 
        val host = args(0)
        val port = args(1)
    
        val configStr =
          s"""
             |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
             |akka.remote.netty.tcp.hostname = "$host"
             |akka.remote.netty.tcp.port = "$port"
          """.stripMargin
        val config: Config = ConfigFactory.parseString(configStr)
    
        //3. 创建actorSystem, akka中的老大, 通过老大才能创建actor
        val sparkMasterSystem = ActorSystem("sparkMasterSystem", config)
    
        //4. 通过老大, 生成actor
        val sparkMaster: ActorRef = sparkMasterSystem.actorOf(Props(new SparkMaster), "sparkMaster")
      }
    }

*  Worker类

import java.util.UUID
    
    import akka.actor.{ Actor, ActorRef, ActorSelection, ActorSystem, Props}
    import com.typesafe.config.{ Config, ConfigFactory}
    
    //定时单位需要导入此包
    import scala.concurrent.duration._
    
    /** * */
    //1. 继承actor
    class SparkWorker(val memory: Int, val cores: Int) extends Actor { 
    
      //定义发送消息的参数
      val workId: String = UUID.randomUUID().toString()
    
      //c. 将master的引用定义到外面, 因为每次发消息都需要这个master引用
      var masterSelection: ActorSelection = _
      //c. 定义定时时间
      val SEND_HEART_BEAT_INTERVAL = 10000 //10s
    
    
      //2. 重写初始化&执行方法
      override def preStart(): Unit = { 
        //需要到sparkMaster中注册,也就是要发消息到主节点, 需要先获取主节点的引用
        masterSelection = context.actorSelection("akka.tcp://sparkMasterSystem@192.168.137.188:8888/user/sparkMaster")
    
        //a. 发送注册消息
        //workId, memory, cores
        masterSelection ! SendRegistMes(workId, memory, cores)
    
      }
    
      //2. 接收消息的方法
      override def receive: Receive = { 
        //b. 接收master发送过来的注册成功的反馈消息
        case RegisteredMsg(msg) => { 
          //1.打印master发送过来的确认消息
          println(msg)
    
          //c. 定时发送心跳-->报活, 需要先发送给自己中转一下
          //定时任务需要导入隐式转换调度方法
          import context.dispatcher
          context.system.scheduler.schedule(0 millis, SEND_HEART_BEAT_INTERVAL millis, self, SendHeartBeat(workId))
    
        }
    
        //c. 接收自己发过来中转的定时消息
        case SendHeartBeat(workerId) => { 
          //d.真正发送心跳给master
          masterSelection ! HeartBeat(workerId)
        }
    
      }
    }
    
    object SparkWorker { 
      def main(args: Array[String]): Unit = { 
        //接收参数
        val host = args(0)
        val port = args(1)
        val memory = args(2).toInt
        val cores = args(3).toInt
    
        val configStr =
          s"""
             |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
             |akka.remote.netty.tcp.hostname = "$host"
             |akka.remote.netty.tcp.port = "$port"
           """.stripMargin
        val config: Config = ConfigFactory.parseString(configStr)
        //3.创建actorSystem老大,这个类是单例, 所以不能new, 用于生产actor
        val sparkWorkerSystem = ActorSystem("sparkWorkerSystem", config)
    
        //4.生产actor
        val sparkWorker: ActorRef = sparkWorkerSystem.actorOf(Props(new SparkWorker(memory, cores)), "sparkWorker")
      }
    }

*  WorkerInfo类

class WorkerInfo(val workerId: String, val memory: Int, val cores: Int) { 
    
      //一定要是var 每次定时心跳会重新赋值
      var time_interval:Long =_
    
      override def toString: String = { 
    // s"workerId=$workerId,memory=$memory,cores=$cores"
        "workerId="+this.workerId+",memory="+this.memory+",cores="+this.cores
      }
    }

*  样例类

trait RemoteMsg extends Serializable{ 
    
    }
    
    //注册, 发送消息的样例类 worker--->master
    case class SendRegistMes(val workerId:String, val memory:Int, val cores:Int) extends RemoteMsg
    
    //master收到注册消息,反馈给worker消息 master--->worker
    case class RegisteredMsg(val msg:String) extends RemoteMsg
    
    //worker定时发送信息 worker--worker 自己发给自己
    case class SendHeartBeat(val workerId:String)
    
    //woker真正发送心跳 worker--master
    case class HeartBeat(val wokerId:String) extends RemoteMsg
    
    //定时检查心跳,因为不需要携带参数,可以使用object
    case object CheckOutTimer

## 2.2 启动&IDEA配置 ##

*  IDEA配置  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 1]  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 2]  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 3]  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 4]
 *  启动：先启动Master，再依次启动多个Worker

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70]: /images/20220126/e19bf8462df74f01830652ce6a8505b2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 1]: /images/20220126/5e993521485a4b0a9fa6f355ce4109d9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 2]: /images/20220126/b49d5ba434bd4f919e67ae78fd6ceba5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 3]: /images/20220126/0e7eadd4facf41dc97a7d525d2d770bc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MjY0MTkwOQ_size_16_color_FFFFFF_t_70 4]: /images/20220126/0ee98d89ad87493591b577d5a973c109.png