基于akka与scala实现一个简单rpc框架

2024-06-02 21:08
文章标签 简单 实现 rpc 框架 scala akka

本文主要是介绍基于akka与scala实现一个简单rpc框架,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

一、RPC简介

RPC,即 Remote Procedure Call(远程过程调用),说得通俗一点就是:调用远程计算机上的服务,就像调用本地服务一样。
RPC 可基于 HTTP 或 TCP 协议,Web Service 就是基于 HTTP 协议的 RPC,它具有良好的跨平台性,但其性能却不如基于 TCP 协议的 RPC。会两方面会直接影响 RPC 的性能,一是传输方式,二是序列化。
众所周知,TCP 是传输层协议,HTTP 是应用层协议,而传输层较应用层更加底层,在数据传输方面,越底层越快,因此,在一般情况下,TCP 一定比 HTTP 快。就序列化而言,Java 提供了默认的序列化方式,但在高并发的情况下,这种方式将会带来一些性能上的瓶颈,于是市面上出现了一系列优秀的序列化框架从而提供更高效的性能。
我们需要将服务部署在分布式环境下的不同节点上,通过服务注册的方式,让客户端来自动发现当前可用的服务,并调用这些服务。这需要一种服务注册表(Service Registry)的组件,让它来注册分布式环境下所有的服务地址(包括:主机名与端口号)。

二、代码框架



三、代码实现

该小项目包含四个文件:

1.WorkerInfo,用于保存Worker的信息,此次保存Worker的上一次心跳时间

package com.zxl.rpcclass WorkerInfo(val id: String, val memory: Int, val cores: Int) {// TODO 上一次心跳var lastHeartbeatTime: Long = _
}
2.RemoteMessage,实现序列化并定义Master与Worker之间传送信息的类型

package com.zxl.rpc/*** 用于实现序列化 网络传输*/
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
3.Worker,与Master进行消息交互

package com.zxl.rpcimport java.util.UUID
import akka.actor.{Props, ActorSystem, Actor, ActorSelection}
import com.typesafe.config.ConfigFactory
import scala.concurrent.duration._class Worker(val masterHost: String, val masterPort: Int, val memory: Int, val cores: Int) extends Actor {// 与master连接的对象var master : ActorSelection = _// 每个worker的idval workerId = UUID.randomUUID().toString// 发送心跳的时间间隔val HEART_INTERVAL = 10000override def preStart(): Unit = {// 建立连接// 在Master启动时会打印下面的那个协议, 可以先用这个做一个标志, 连接哪个master// 继承actor后会有一个context, 可以通过它来连接// 需要有/user, Master要和master那边创建的名字保持一致master = context.actorSelection(s"akka.tcp://MasterSystem@$masterHost:$masterPort/user/Master")// 向Master发送注册消息master ! RegisterWorker(workerId, memory, cores)}override def receive: Receive = {case RegisteredWorker(masterUrl) => {println(masterUrl)// 启动定时器发送心跳import context.dispatcher// 多长时间后执行 单位,多长时间执行一次 单位, 消息的接受者(直接给master发不好, 先给自己发送消息, 以后可以做下判断, 什么情况下再发送消息), 信息context.system.scheduler.schedule(0 millis, HEART_INTERVAL millis, self, SendHeartbeat)}case SendHeartbeat => {println("send heartbeat to master")// 对master发送心跳信息,发送当前worker的idmaster ! Heartbeat(workerId)}}
}object Worker {def main(args: Array[String]) {val host = args(0)val port = args(1).toIntval 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"""".stripMarginval config = ConfigFactory.parseString(configStr)// ActorSystem老大,辅助创建和监控下面的Actor,它是单例的val actorSystem = ActorSystem("WorkerSystem", config)actorSystem.actorOf(Props(new Worker(masterHost, masterPort, memory, cores)), "Worker")actorSystem.awaitTermination()}
}
4.Master,接收Worker的消息并做回应

package com.zxl.rpcimport akka.actor.{Props, ActorSystem, Actor}
import com.typesafe.config.ConfigFactory
import scala.collection.mutable
import scala.concurrent.duration._class Master(val host: String, val port: Int) extends Actor {// 保存(workerId,WorkerInfo)val idToWorker = new mutable.HashMap[String, WorkerInfo]()// 保存wokerInfo// 使用set删除快, 也可用linkListval workers = new mutable.HashSet[WorkerInfo]()// 超时检查的间隔val CHECK_INTERVAL = 15000override def preStart(): Unit = {println("preStart invoked")// 导入隐式转换import context.dispatcher// 使用timer太low了, 可以使用akka的, 使用定时器, 要导入这个包// 定时检查worker的心跳时间是否超时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) = workerInfoworkers += workerInfo// 通知worker注册 并将master的地址返回给workersender ! RegisteredWorker(s"akka.tcp://MasterSystem@$host:$port/user/Master")}}case Heartbeat(id) => {if(idToWorker.contains(id)) {val workerInfo = idToWorker(id)// 报活val currentTime = System.currentTimeMillis()// 更新已收到id的worker的上一次心跳时间workerInfo.lastHeartbeatTime = currentTime}}case CheckTimeOutWorker => {val currentTime = System.currentTimeMillis()// 获取过时的workerval toRemove = workers.filter(x => currentTime - x.lastHeartbeatTime > CHECK_INTERVAL)// 将超时的worker从两个集合中去掉for(w <- toRemove) {workers -= widToWorker -= w.id}println(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"""".stripMarginval config = ConfigFactory.parseString(configStr)// ActorSystem老大,辅助创建和监控下面的Actor,它是单例的val actorSystem = ActorSystem("MasterSystem", config)// 创建Actorval master = actorSystem.actorOf(Props(new Master(host, port)), "Master")master ! "hello"actorSystem.awaitTermination()}
}


这篇关于基于akka与scala实现一个简单rpc框架的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1025087

相关文章

Java的栈与队列实现代码解析

《Java的栈与队列实现代码解析》栈是常见的线性数据结构,栈的特点是以先进后出的形式,后进先出,先进后出,分为栈底和栈顶,栈应用于内存的分配,表达式求值,存储临时的数据和方法的调用等,本文给大家介绍J... 目录栈的概念(Stack)栈的实现代码队列(Queue)模拟实现队列(双链表实现)循环队列(循环数组

C++如何通过Qt反射机制实现数据类序列化

《C++如何通过Qt反射机制实现数据类序列化》在C++工程中经常需要使用数据类,并对数据类进行存储、打印、调试等操作,所以本文就来聊聊C++如何通过Qt反射机制实现数据类序列化吧... 目录设计预期设计思路代码实现使用方法在 C++ 工程中经常需要使用数据类,并对数据类进行存储、打印、调试等操作。由于数据类

Python实现图片分割的多种方法总结

《Python实现图片分割的多种方法总结》图片分割是图像处理中的一个重要任务,它的目标是将图像划分为多个区域或者对象,本文为大家整理了一些常用的分割方法,大家可以根据需求自行选择... 目录1. 基于传统图像处理的分割方法(1) 使用固定阈值分割图片(2) 自适应阈值分割(3) 使用图像边缘检测分割(4)

Android实现在线预览office文档的示例详解

《Android实现在线预览office文档的示例详解》在移动端展示在线Office文档(如Word、Excel、PPT)是一项常见需求,这篇文章为大家重点介绍了两种方案的实现方法,希望对大家有一定的... 目录一、项目概述二、相关技术知识三、实现思路3.1 方案一:WebView + Office Onl

C# foreach 循环中获取索引的实现方式

《C#foreach循环中获取索引的实现方式》:本文主要介绍C#foreach循环中获取索引的实现方式,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录一、手动维护索引变量二、LINQ Select + 元组解构三、扩展方法封装索引四、使用 for 循环替代

Spring Security+JWT如何实现前后端分离权限控制

《SpringSecurity+JWT如何实现前后端分离权限控制》本篇将手把手教你用SpringSecurity+JWT搭建一套完整的登录认证与权限控制体系,具有很好的参考价值,希望对大家... 目录Spring Security+JWT实现前后端分离权限控制实战一、为什么要用 JWT?二、JWT 基本结构

Java实现优雅日期处理的方案详解

《Java实现优雅日期处理的方案详解》在我们的日常工作中,需要经常处理各种格式,各种类似的的日期或者时间,下面我们就来看看如何使用java处理这样的日期问题吧,感兴趣的小伙伴可以跟随小编一起学习一下... 目录前言一、日期的坑1.1 日期格式化陷阱1.2 时区转换二、优雅方案的进阶之路2.1 线程安全重构2

Android实现两台手机屏幕共享和远程控制功能

《Android实现两台手机屏幕共享和远程控制功能》在远程协助、在线教学、技术支持等多种场景下,实时获得另一部移动设备的屏幕画面,并对其进行操作,具有极高的应用价值,本项目旨在实现两台Android手... 目录一、项目概述二、相关知识2.1 MediaProjection API2.2 Socket 网络

使用Python实现图像LBP特征提取的操作方法

《使用Python实现图像LBP特征提取的操作方法》LBP特征叫做局部二值模式,常用于纹理特征提取,并在纹理分类中具有较强的区分能力,本文给大家介绍了如何使用Python实现图像LBP特征提取的操作方... 目录一、LBP特征介绍二、LBP特征描述三、一些改进版本的LBP1.圆形LBP算子2.旋转不变的LB

Redis消息队列实现异步秒杀功能

《Redis消息队列实现异步秒杀功能》在高并发场景下,为了提高秒杀业务的性能,可将部分工作交给Redis处理,并通过异步方式执行,Redis提供了多种数据结构来实现消息队列,总结三种,本文详细介绍Re... 目录1 Redis消息队列1.1 List 结构1.2 Pub/Sub 模式1.3 Stream 结