本文主要是介绍Scala语法(六) Akka与线程通信，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！
前言
在初期, Scala可以通过Akka来实现线程通信. 当然, 现在还支持使用Netty方式进行通信.
本章主要介绍使用Akka方式进行通信的写法.
正文
- Master结点
import akka.actor.Actor
import akka.actor.ActorSystem
import com.typesafe.config.ConfigFactory
import akka.actor.Propsclass AkkaMaster extends Actor{// start 之前override def preStart() : Unit = {println("pre master invoke.")}// 用于接收消息override def receive:Receive = {case "connect" => {println("a client connected.")sender ! "reply"}case "hello" => {println("hello")}}
}
object AkkaMaster{def main(args: Array[String]): Unit = {// 使用创建ActorSystem来创建和监控下面的Actor对象. 单例的.val host = "127.0.0.1"val port = 8090// 准备配置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)// 注意名称中间不要加空格val actorSysetm = ActorSystem("MasterSystem",config) // 创建Actorval master = actorSysetm.actorOf(Props(new AkkaMaster),"Master")master ! "hello"// 等待信号停止actorSysetm.awaitTermination()}
}// 顺利输出
//[INFO] [04/29/2019 16:43:20.512] [main] [Remoting] Starting remoting
//[INFO] [04/29/2019 16:43:20.770] [main] [Remoting] Remoting started; listening on addresses :[akka.tcp://MasterSystem@127.0.0.1:8090]
//[INFO] [04/29/2019 16:43:20.771] [main] [Remoting] Remoting now listens on addresses: [akka.tcp://MasterSystem@127.0.0.1:8090]
//pre master invoke.
//hello// 1. 名称中间不要加空格
//Exception in thread "main" java.lang.IllegalArgumentException: invalid ActorSystem name [Master System], must contain only word characters (i.e. [a-zA-Z0-9] plus non-leading '-' or '_')
// at akka.actor.ActorSystemImpl.<init>(ActorSystem.scala:498)
// at akka.actor.ActorSystem$.apply(ActorSystem.scala:142)
// at akka.actor.ActorSystem$.apply(ActorSystem.scala:119)
// at com.yanxml.quick_scala.multi.akka.AkkaMaster$.main(AkkaMaster.scala:33)
// at com.yanxml.quick_scala.multi.akka.AkkaMaster.main(AkkaMaster.scala)
- Worker结点
import akka.actor.Actor
import akka.actor.ActorSelection
import akka.actor.ActorSystem
import akka.actor.Props
import com.typesafe.config.ConfigFactoryclass AkkaWorker extends Actor{// 成员变量val master:ActorSelection = null// 建立链接override def preStart():Unit = {val master = context.actorSelection("akka.tcp://MasterSystem@127.0.0.1:8090/user/Master")println(master)master ! "connect"}override def receive:Receive = {case "reply" => {println("a reply from master")}}
}object AkkaWorker{def main(args: Array[String]): Unit = {// 使用创建ActorSystem来创建和监控下面的Actor对象. 单例的.val host = "127.0.0.1"val port = 8091// 准备配置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)// 注意名称中间不要加空格val actorSysetm = ActorSystem("WorkerSystem",config) // 创建Actorval master = actorSysetm.actorOf(Props(new AkkaWorker),"Worker")master ! "hello"// 等待信号停止actorSysetm.awaitTermination()}
}
模拟RPC
在模拟RPC中主要有这样的流程.
其中主要包括两个结点: Worker结点&Master结点.
- 运行流程:
Master结点先进行启动.Worker结点后进行启动.Worker结点向Master结点发送注册消息.Matser结点接收注册消息, 并进行记录. 并将主结点的地址返回给Worker结点(模拟Master是集群的情况).并记录,最后的通信时间作为心跳标志.Worker结点接收主结点地址, 并形成通信链接. 开始通信. 并定时发送心跳消息.
改造上方的Demo代码. 其基本代码如下所示:
- RemoteMessage
trait RemoteMessage extends Serializable{}// Worker -> Master 用来封装Worker信息
case class RegisterWorker(id:String,memory:Int,cores:Int)class WorkerInfo(val id:String, val memory:Int, val cores:Int){// 上一次心跳var heartbeatTime:String = _
}
- Master
import akka.actor.Actor
import akka.actor.ActorSystem
import com.typesafe.config.ConfigFactory
import akka.actor.Props
import scala.collection.immutable.HashMapprivate [simulate] class AkkaMaster extends Actor{val idToWorker = new scala.collection.mutable.HashMap[String,WorkerInfo]()// start 之前override def preStart() : Unit = {println("pre master invoke.")}// 用于接收消息override def receive:Receive = {case "connect" => {println("a client connected.")sender ! "reply"}case "hello" => {println("hello")}// 传输样例类case RegisterWorker(id,memory,cores)=>{if(!idToWorker.contains(id)){idToWorker.put(id, new WorkerInfo(id,memory,cores))}sender ! "123"}}
}
object AkkaMaster{def main(args: Array[String]): Unit = {// 使用创建ActorSystem来创建和监控下面的Actor对象. 单例的.val host = "127.0.0.1"val port = 8090// 准备配置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)// 注意名称中间不要加空格val actorSysetm = ActorSystem("MasterSystem",config) // 创建Actorval master = actorSysetm.actorOf(Props(new AkkaMaster),"Master")master ! "hello"// 等待信号停止actorSysetm.awaitTermination()}
}// 顺利输出
//[INFO] [04/29/2019 16:43:20.512] [main] [Remoting] Starting remoting
//[INFO] [04/29/2019 16:43:20.770] [main] [Remoting] Remoting started; listening on addresses :[akka.tcp://MasterSystem@127.0.0.1:8090]
//[INFO] [04/29/2019 16:43:20.771] [main] [Remoting] Remoting now listens on addresses: [akka.tcp://MasterSystem@127.0.0.1:8090]
//pre master invoke.
//hello// 1. 名称中间不要加空格
//Exception in thread "main" java.lang.IllegalArgumentException: invalid ActorSystem name [Master System], must contain only word characters (i.e. [a-zA-Z0-9] plus non-leading '-' or '_')
// at akka.actor.ActorSystemImpl.<init>(ActorSystem.scala:498)
// at akka.actor.ActorSystem$.apply(ActorSystem.scala:142)
// at akka.actor.ActorSystem$.apply(ActorSystem.scala:119)
// at com.yanxml.quick_scala.multi.akka.AkkaMaster$.main(AkkaMaster.scala:33)
// at com.yanxml.quick_scala.multi.akka.AkkaMaster.main(AkkaMaster.scala)- Worker
import akka.actor.Actor
import akka.actor.ActorSelection
import akka.actor.ActorSystem
import akka.actor.Props
import com.typesafe.config.ConfigFactory
import com.yanxml.quick_scala.multi.akka.simulate.RegisterWorker
import java.util.UUIDprivate [simulate] class AkkaWorker(val masterHost:String, val masterPort:String, val memory:Int, val cores:Int) extends Actor{// 成员变量val master:ActorSelection = null// 建立链接override def preStart():Unit = {// 和Master建立链接val master = context.actorSelection(s"akka.tcp://MasterSystem@$masterHost:$masterPort/user/Master")val workerId = UUID.randomUUID().toString()//println(master)// 向Master发送消息master ! RegisterWorker(workerId,memory,cores)}override def receive:Receive = {case "reply" => {println("a reply from master")}}
}object AkkaWorker{def main(args: Array[String]): Unit = {// 使用创建ActorSystem来创建和监控下面的Actor对象. 单例的.val host = "127.0.0.1"val port = 8091// 准备配置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)// 注意名称中间不要加空格val actorSysetm = ActorSystem("WorkerSystem",config) // 创建Actorval master = actorSysetm.actorOf(Props(new AkkaWorker("127.0.0.1","8090",2,2)),"Worker")master ! "hello"// 等待信号停止actorSysetm.awaitTermination()}
}注: 后续的通信逻辑就是丰富双方的receive()方法.
这篇关于Scala语法(六) Akka与线程通信的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！
