TOP

服务注册、发现、心跳(一)
2020-05-21 06:12:39 】 浏览:17次 本网站的内容取自网络,仅供学习参考之用,绝无侵犯任何人知识产权之意。如有侵犯请您及时与本人取得联系,万分感谢。
Tags:服务 注册 发现 心跳

使用 akka 实现,服务注册、发现、心跳

消息类:

package org.example

case class RegisterWorkerInfo(id: String, cpu: Int, ram: Int);

case class WorkerInfo(val id: String, val cpu: Int, val ram: Int) {
var lastheartbeat: Long = System.currentTimeMillis();
};

case object RegisteredWorkerInfo;

case object SendHeartBeat;

case object StartTimeOutWorker;

case object RemoveTimeOutWorker;

case class HeartBeanPackage(id: String);

服务端 master:
package org.example

import java.util.concurrent.TimeUnit

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

import scala.collection.mutable
import scala.concurrent.duration.FiniteDuration

class MasterActor extends Actor {
val register = mutable.Map[String, WorkerInfo]();

override def receive: Receive = {
case "start" => {
println("master start !!!");
self ! StartTimeOutWorker;
}
case RegisterWorkerInfo(id, cpu, ram) => {
if (!register.contains(id)) {
val workerInfo = WorkerInfo(id, cpu, ram);
register(id) = workerInfo;
};

println("workers: " + register)
sender() ! RegisteredWorkerInfo;
}
case HeartBeanPackage(id) => {
val workerInfo: WorkerInfo = register(id);
workerInfo.lastheartbeat = System.currentTimeMillis();
println("master 更新了 " + id + "心跳时间");
}
case StartTimeOutWorker =>{
import context.dispatcher;
context.system.scheduler.schedule(FiniteDuration(0, TimeUnit.SECONDS), FiniteDuration(9, TimeUnit.SECONDS),
self, RemoveTimeOutWorker);
}
case RemoveTimeOutWorker =>{
register.filter(x => System.currentTimeMillis() - x._2.lastheartbeat > 6000)
.foreach(x => register.remove(x._1));
println("在线 worker数量: " + register.size );
}
}
}

object MasterActor {
def main(args: Array[String]): Unit = {
val config = ConfigFactory.parseString(
s"""
|akka.actor.provider="akka.remote.RemoteActorRefProvider"
|akka.remote.netty.tcp.hostname=localhost
|akka.remote.netty.tcp.port=8080
""".stripMargin);
val masterSysterm = ActorSystem("master_systerm", config);
val masterActor1 = masterSysterm.actorOf(Props[MasterActor], "master_actor1")
masterActor1 ! "start";
}
}

客户端 slaver:
package org.example

import java.util.UUID
import java.util.concurrent.TimeUnit

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

import scala.concurrent.duration.{FiniteDuration, TimeUnit}

class SlaveActor extends Actor {
var masterProxy: ActorSelection = _;
val id: String = UUID.randomUUID().toString;

override def preStart(): Unit = {
val masterUrl = "akka.tcp://master_systerm@localhost:8080/user/master_actor1";
this.masterProxy = context.actorSelection(masterUrl);
}


override def receive: Receive = {
case "start" => {
println("客户端运行了");
this.masterProxy ! RegisterWorkerInfo(id, 16, 16 * 1024);
}
case RegisteredWorkerInfo => {
println("workerId: " + id + " 注册成功");

import context.dispatcher;
context.system.scheduler.schedule(FiniteDuration(0, TimeUnit.SECONDS), FiniteDuration(3, TimeUnit.SECONDS), self, SendHeartBeat);
}
case SendHeartBeat =>{
println("workerId: " + id + " 给服务器发送心跳");
masterProxy ! HeartBeanPackage(id);
}
}
}

object SlaveActor {
def main(args: Array[String]

请关注公众号获取更多资料


服务注册、发现、心跳(一) https://www.cppentry.com/bencandy.php?fid=97&id=289200

首页 上一页 1 2 下一页 尾页 1/2/2
】【打印繁体】【投稿】【收藏】 【推荐】【举报】【评论】 【关闭】 【返回顶部
上一篇从零开始搭建一个PaaS平台 - 我们.. 下一篇北航OO(2020)第三单元博客作业

评论

验 证 码:
表  情:
内  容: