手写一个 Mini Flink 分布式集群

package com.mazh.rpc.akka.flink

import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.ConfigFactory

import scala.collection.mutable


class MyJobManager(var hostname: String, var port: Int) extends Actor {
    
    // 用来存储每个注册的 TaskManager 节点的信息
    private var id2taskManagerInfo = new mutable.HashMap[String, TaskManagerInfo]()
    
    // 对所有注册的 TaskManager 进行去重，其实就是一个 HashSet
    private var taskManagerInfos = new mutable.HashSet[TaskManagerInfo]()
    
    //actor 在最开始的时候，会执行一次
    override def preStart(): Unit = {
        import scala.concurrent.duration._
        import context.dispatcher
        
        // 调度一个任务， 每隔五秒钟执行一次
        context.system.scheduler.schedule(0 millis, 5000 millis, self, CheckTimeOut)
    }
    
    // 正经服务方法
    override def receive: Receive = {
        
        //  接收 注册消息
        case RegisterTaskManager(taskmanagerid, memory, cpu) => {
            val taskManagerInfo = new TaskManagerInfo(taskmanagerid, memory, cpu)
            println(s"节点 ${taskmanagerid} 上线")
            
            //  对注册的 TaskManager 节点进行存储管理
            id2taskManagerInfo.put(taskmanagerid, taskManagerInfo)
            taskManagerInfos += taskManagerInfo
            
            //  把信息存到zookeeper
            //  sender() 谁给我发消息，sender方法返回的就是谁
            sender() ! RegisteredTaskManager(hostname + ":" + port)
        }
        
            // 接收心跳消息
        case Heartbeat(taskManagerId) => {
            val currentTime = System.currentTimeMillis()
            val taskManagerInfo = id2taskManagerInfo(taskManagerId)
            taskManagerInfo.lastHeartBeatTime = currentTime
    
            id2taskManagerInfo(taskManagerId) = taskManagerInfo
            taskManagerInfos += taskManagerInfo
        }
        
        //  检查过期失效的 NodeManager
        case CheckTimeOut => {
            val currentTime = System.currentTimeMillis()
            
            // TODO_MA 注释：15 秒钟失效
            taskManagerInfos.filter(nm => {
                val heartbeatTimeout = 15000
                val bool = currentTime - nm.lastHeartBeatTime > heartbeatTimeout
                if (bool) {
                    println(s"节点 ${nm.taskmanagerid} 下线")
                }
                bool
            }).foreach(deadnm => {
                taskManagerInfos -= deadnm
                id2taskManagerInfo.remove(deadnm.taskmanagerid)
            })
            println("当前注册成功的节点数" + taskManagerInfos.size + "\t分别是：" + taskManagerInfos.map(x => x.toString)
              .mkString(","));
        }
    }
}


object MyJobManager {
    def main(args: Array[String]): Unit = {
        
        //  地址参数
        val str =
        s"""
           |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
           |akka.remote.netty.tcp.hostname = localhost
           |akka.remote.netty.tcp.port = 6789
      """.stripMargin
        val conf = ConfigFactory.parseString(str)
        
        // ActorSystem
        val actorSystem = ActorSystem(Constant.JMAS, conf)
        
        // 启动了一个actor ：MyJobManager
        actorSystem.actorOf(Props(new MyJobManager("localhost", 6789)), Constant.JMA)
   
    }
}

package com.mazh.rpc.akka.flink

import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import com.typesafe.config.ConfigFactory


class MyTaskManager(val tmhostname: String, val jobmanagerhostname: String, val jobmanagerport: Int, val memory: Int,
    val cpu: Int) extends Actor {
    
    var taskManagerId: String = tmhostname
    var rmRef: ActorSelection = _
    
    // 会提前执行一次
    // 当前NM启动好了之后，就应该给 RM 发送一个注册消息
    //发给谁，就需要获取这个谁的一个ref实例
    override def preStart(): Unit = {
        
        //获取消息发送对象的一个ref实例
        // 远程path  akka.tcp://（ActorSystem的名称）@（远程地址的IP）：（远程地址的端口）/user/（Actor的名称）
        rmRef = context.actorSelection(s"akka.tcp://${
            Constant.JMAS
        }@${jobmanagerhostname}:${jobmanagerport}/user/${Constant.JMA}")
        
        //  发送消息
        println(taskManagerId + " 正在注册")
        rmRef ! RegisterTaskManager(taskManagerId, memory, cpu)
    }
    
    //正常服务方法
    override def receive: Receive = {
        
        //接收到注册成功的消息
        case RegisteredTaskManager(masterURL) => {
            println(masterURL);
            
            //  initialDelay: FiniteDuration, 多久以后开始执行
            //  interval:     FiniteDuration, 每隔多长时间执行一次
            // receiver:     ActorRef, 给谁发送这个消息
            //  message:      Any  发送的消息是啥
            import scala.concurrent.duration._
            import context.dispatcher
            context.system.scheduler.schedule(0 millis, 4000 millis, self, SendMessage)
        }
        
        // 发送心跳
        case SendMessage => {
            // TODO_MA 注释： 向主节点发送心跳信息
            rmRef ! Heartbeat(taskManagerId)
            
            println(Thread.currentThread().getId)
        }
    }
}


object MyTaskManager {
    def main(args: Array[String]): Unit = {
        
        // 远程主机名称
        val HOSTNAME = args(0)
        
        //   JobManager 的 hostname 和 port
        val JM_HOSTNAME = args(1)
        val JM_PORT = args(2).toInt
        
        //  抽象的内存资源 和 CPU 个数
        val TaskManager_MEMORY = args(3).toInt
        val TaskManager_CORE = args(4).toInt
        
        //  当前 TaskManager 的 hostname 和 port
        var TaskManager_PORT = args(5).toInt
        var TaskManager_HOSTNAME = args(6)
        
        //   指定主机名称和端口号相关的配置
        val str =
            s"""
               |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
               |akka.remote.netty.tcp.hostname = ${HOSTNAME}
               |akka.remote.netty.tcp.port = ${TaskManager_PORT}
            """.stripMargin
        val conf = ConfigFactory.parseString(str)
        
        // 启动一个 ActorSystem
        val actorSystem = ActorSystem(Constant.JMAS, conf)
        
        //  启动一个Actor
        actorSystem.actorOf(Props(new MyTaskManager(TaskManager_HOSTNAME, JM_HOSTNAME, JM_PORT, TaskManager_MEMORY,
            TaskManager_CORE)), Constant.TMA)
    }
}