用AKKA实现简单的RPC通信模型2
2017-06-16 22:07
429 查看
上一篇文章,简单地用AKKA实现了RPC通信,这篇文章在之前的基础之上,进行了更多地关于RPC通信的操作,包括传一些相关数据,包括心跳机制等
具体功能如下图:
代码结构:
WorkerInfo.scala代码
package cn.heres.rpc
/**
* Created by vinsuan on 2017/6/15 0015.
*/
class WorkerInfo(val id: String,val memory:Int,val cores:Int) {
//todo 上一次心跳
var lastHeartBeatTime : Long = _
}
RemoteMessage.scala代码
package cn.heres.rpc
/**
* Created by vinsuan on 2017/6/15 0015.
*/
trait RemoteMessage extends Serializable
//worker->Master
case class RegisterWorker(id:String , memory : Int ,cores:Int) extends RemoteMessage
case class Heartbeat(id:String ) extends RemoteMessage
//Master->worker
case class RegisteredWorker(masterUrl: String) extends RemoteMessage
//worker->self
case object SendHeartbeat
//Master->self
case object CheckTimeOutWorker
Worker.scala代码:
package cn.heres.rpc
import java.util.UUID
import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import akka.actor.Actor.Receive
import com.typesafe.config.ConfigFactory
import scala.concurrent.duration._
/**
* Created by vinsuan on 2017/6/13 0013.
*/
class Worker(val masterHost : String ,val masterPost : Int,val memory: Int ,val cores :Int) extends Actor{
var master : ActorSelection = _
val workerId = UUID.randomUUID().toString
val HEARTBEAT_INTERVAL = 10000
//建立连接
override def preStart(): Unit = {
master = context.actorSelection(s"akka.tcp://MasterSystem@$masterHost:$masterPost/user/Master")
master ! RegisterWorker(workerId,memory,cores)
}
override def receive: Receive = {
case RegisteredWorker(masterUrl) => {
println(masterUrl)
//导入隐式转换
import context.dispatcher
//启动定时器发送心跳
context.system.scheduler.schedule(0 millis,HEARTBEAT_INTERVAL millis,self,SendHeartbeat)
}
case SendHeartbeat => {
println("send heartbeat to master")
master ! Heartbeat(workerId)
}
}
}
object Worker{
def main(args: Array[String]) {
val host = args(0)
val port = args(1).toInt
val masterHost = args(2)
val masterPort = args(3).toInt
val memory = args(4).toInt
val cores = args(5).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 = ConfigFactory.parseString(configStr)
//ActorSystem 老大,辅助创建和监控下面的Actor ,它是单例的
val actorSystem = ActorSystem("workerSystem",config)
//创建Actor
val worker = actorSystem.actorOf(Props(new Worker(masterHost,masterPort,memory,cores)),"Worker")
actorSystem.awaitTermination()
}
}
Maser.scala代码:
package cn.heres.rpc
import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
import scala.concurrent.duration._
/**
* Created by vinsuan on 2017/6/13 0013.
*/
class Master(val host:String,val port:Int) extends Actor {
//workerId -> workerInfo
val idtoWorker = new scala.collection.mutable.HashMap[String, WorkerInfo]()
//workerInfo
val workers = new scala.collection.mutable.HashSet[WorkerInfo]()
val CHECK_INTERVAL = 15000
//超时检测的间隔
override def preStart(): Unit = {
println("preStart invoked")
//导入隐式转换
import context.dispatcher
context.system.scheduler.schedule(0 millis, CHECK_INTERVAL millis, self, CheckTimeOutWorker)
}
//用于接收消息
override def receive: Receive = {
case RegisterWorker(id, memory, cores) => {
//判断一下,是不是已经注册过
if (!idtoWorker.contains(id)) {
//把Worker的信息封装起来保存到内存中
val workerInfo = new WorkerInfo(id, memory, cores)
idtoWorker(id) = workerInfo
workers += workerInfo
sender ! RegisteredWorker(s"akka.tcp://MasterSystem@$host:$port/user/Master")
}
}
case Heartbeat(id) => {
if (idtoWorker.contains(id)) {
val workerInfo = idtoWorker(id)
//报活
val currentTime = System.currentTimeMillis()
workerInfo.lastHeartBeatTime = currentTime
}
}
case CheckTimeOutWorker => {
val currentTime = System.currentTimeMillis()
val toRemove = workers.filter(x => currentTime - x.lastHeartBeatTime > CHECK_INTERVAL)
for (w <- toRemove) {
workers -= w
idtoWorker -= w.id
}
println("还有几个活着的worker:" + workers.size)
}
}
}
object Master {
def main(args: Array[String]) {
val host = args(0)
val port = args(1).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 = ConfigFactory.parseString(configStr)
//ActorSystem 老大,辅助创建和监控下面的Actor ,它是单例的
val actorSystem = ActorSystem("MasterSystem", config)
//创建Actor
val master = actorSystem.actorOf(Props(new Master(host, port)), "Master")
actorSystem.awaitTermination()
}
}
具体功能如下图:
代码结构:
WorkerInfo.scala代码
package cn.heres.rpc
/**
* Created by vinsuan on 2017/6/15 0015.
*/
class WorkerInfo(val id: String,val memory:Int,val cores:Int) {
//todo 上一次心跳
var lastHeartBeatTime : Long = _
}
RemoteMessage.scala代码
package cn.heres.rpc
/**
* Created by vinsuan on 2017/6/15 0015.
*/
trait RemoteMessage extends Serializable
//worker->Master
case class RegisterWorker(id:String , memory : Int ,cores:Int) extends RemoteMessage
case class Heartbeat(id:String ) extends RemoteMessage
//Master->worker
case class RegisteredWorker(masterUrl: String) extends RemoteMessage
//worker->self
case object SendHeartbeat
//Master->self
case object CheckTimeOutWorker
Worker.scala代码:
package cn.heres.rpc
import java.util.UUID
import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import akka.actor.Actor.Receive
import com.typesafe.config.ConfigFactory
import scala.concurrent.duration._
/**
* Created by vinsuan on 2017/6/13 0013.
*/
class Worker(val masterHost : String ,val masterPost : Int,val memory: Int ,val cores :Int) extends Actor{
var master : ActorSelection = _
val workerId = UUID.randomUUID().toString
val HEARTBEAT_INTERVAL = 10000
//建立连接
override def preStart(): Unit = {
master = context.actorSelection(s"akka.tcp://MasterSystem@$masterHost:$masterPost/user/Master")
master ! RegisterWorker(workerId,memory,cores)
}
override def receive: Receive = {
case RegisteredWorker(masterUrl) => {
println(masterUrl)
//导入隐式转换
import context.dispatcher
//启动定时器发送心跳
context.system.scheduler.schedule(0 millis,HEARTBEAT_INTERVAL millis,self,SendHeartbeat)
}
case SendHeartbeat => {
println("send heartbeat to master")
master ! Heartbeat(workerId)
}
}
}
object Worker{
def main(args: Array[String]) {
val host = args(0)
val port = args(1).toInt
val masterHost = args(2)
val masterPort = args(3).toInt
val memory = args(4).toInt
val cores = args(5).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 = ConfigFactory.parseString(configStr)
//ActorSystem 老大,辅助创建和监控下面的Actor ,它是单例的
val actorSystem = ActorSystem("workerSystem",config)
//创建Actor
val worker = actorSystem.actorOf(Props(new Worker(masterHost,masterPort,memory,cores)),"Worker")
actorSystem.awaitTermination()
}
}
Maser.scala代码:
package cn.heres.rpc
import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
import scala.concurrent.duration._
/**
* Created by vinsuan on 2017/6/13 0013.
*/
class Master(val host:String,val port:Int) extends Actor {
//workerId -> workerInfo
val idtoWorker = new scala.collection.mutable.HashMap[String, WorkerInfo]()
//workerInfo
val workers = new scala.collection.mutable.HashSet[WorkerInfo]()
val CHECK_INTERVAL = 15000
//超时检测的间隔
override def preStart(): Unit = {
println("preStart invoked")
//导入隐式转换
import context.dispatcher
context.system.scheduler.schedule(0 millis, CHECK_INTERVAL millis, self, CheckTimeOutWorker)
}
//用于接收消息
override def receive: Receive = {
case RegisterWorker(id, memory, cores) => {
//判断一下,是不是已经注册过
if (!idtoWorker.contains(id)) {
//把Worker的信息封装起来保存到内存中
val workerInfo = new WorkerInfo(id, memory, cores)
idtoWorker(id) = workerInfo
workers += workerInfo
sender ! RegisteredWorker(s"akka.tcp://MasterSystem@$host:$port/user/Master")
}
}
case Heartbeat(id) => {
if (idtoWorker.contains(id)) {
val workerInfo = idtoWorker(id)
//报活
val currentTime = System.currentTimeMillis()
workerInfo.lastHeartBeatTime = currentTime
}
}
case CheckTimeOutWorker => {
val currentTime = System.currentTimeMillis()
val toRemove = workers.filter(x => currentTime - x.lastHeartBeatTime > CHECK_INTERVAL)
for (w <- toRemove) {
workers -= w
idtoWorker -= w.id
}
println("还有几个活着的worker:" + workers.size)
}
}
}
object Master {
def main(args: Array[String]) {
val host = args(0)
val port = args(1).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 = ConfigFactory.parseString(configStr)
//ActorSystem 老大,辅助创建和监控下面的Actor ,它是单例的
val actorSystem = ActorSystem("MasterSystem", config)
//创建Actor
val master = actorSystem.actorOf(Props(new Master(host, port)), "Master")
actorSystem.awaitTermination()
}
}
相关文章推荐
- 用Java socket (TCP通信模型)实现一个简单的web 服务器
- scala-akka实现简单的分布式RPC通信框架(主从监听,消息发送)
- 【远程调用框架】如何实现一个简单的RPC框架(四)优化二:改变底层通信框架
- 基于akka与scala实现一个简单rpc框架
- 分布式Web应用----基于Socket+动态代理实现简单RPC 生产者消费者模型
- Scala基于Akka的Remote Actor实现的简单RPC
- 使用Akka实现一个简单的RPC框架(一)
- 使用Akka实现一个简单的RPC框架(二)
- 简单实现gSOAP和ASP.NET的通信
- linux下socket实现TCP通信的简单程序接口封装
- 用VB实现程序之间的简单通信
- 用多路复用I/O模型实现支持多个客户端的通信服务
- C++ 简单的 Tcp 实现[socket] 客户端与客户端通信
- 即时通信的简单实现
- C# 网络编程(1)——C# 简单实现局域网内通信
- 使用正向最大匹配算法实现中文分词简单模型-用trie树实现
- Java 利用套接字Socket实现简单的服务器与客户端通信
- 用多路复用I/O模型实现支持多个客户端的通信服务
- 快速实现一个简单的bigpipe模型
- Remoting技术初探--实现简单服务器客户机通信