Scala语法(六) Akka与线程通信

2024-05-14 07:58
文章标签 线程 语法 通信 scala akka

本文主要是介绍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与线程通信的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/988198。

相关文章

Python正则表达式语法及re模块中的常用函数详解

Python正则表达式语法及re模块中的常用函数详解

《Python正则表达式语法及re模块中的常用函数详解》这篇文章主要给大家介绍了关于Python正则表达式语法及re模块中常用函数的相关资料,正则表达式是一种强大的字符串处理工具,可以用于匹配、切分、... 目录概念、作用和步骤语法re模块中的常用函数总结 概念、作用和步骤概念： 本身也是一个字符串，其中
阅读更多...
JAVA保证HashMap线程安全的几种方式

JAVA保证HashMap线程安全的几种方式

《JAVA保证HashMap线程安全的几种方式》HashMap是线程不安全的,这意味着如果多个线程并发地访问和修改同一个HashMap实例,可能会导致数据不一致和其他线程安全问题,本文主要介绍了JAV... 目录1. 使用 Collections.synchronizedMap2. 使用 Concurren
阅读更多...
Mysql用户授权(GRANT)语法及示例解读

Mysql用户授权(GRANT)语法及示例解读

《Mysql用户授权(GRANT)语法及示例解读》：本文主要介绍Mysql用户授权(GRANT)语法及示例,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录mysql用户授权（GRANT）语法授予用户权限语法GRANT语句中的<权限类型>的使用WITH GRANT
阅读更多...
HTML5表格语法格式详解

HTML5表格语法格式详解

《HTML5表格语法格式详解》在HTML语法中，表格主要通过table、tr和td3个标签构成,本文通过实例代码讲解HTML5表格语法格式，感兴趣的朋友一起看看吧... 目录一、表格1.表格语法格式2.表格属性 3.例子二、不规则表格1.跨行2.跨列3.例子一、表格在html语法中，表格主要通过< tab
阅读更多...
Java使用ANTLR4对Lua脚本语法校验详解

Java使用ANTLR4对Lua脚本语法校验详解

《Java使用ANTLR4对Lua脚本语法校验详解》ANTLR是一个强大的解析器生成器,用于读取、处理、执行或翻译结构化文本或二进制文件,下面就跟随小编一起看看Java如何使用ANTLR4对Lua脚本... 目录什么是ANTLR？第一个例子ANTLR4 的工作流程Lua脚本语法校验准备一个Lua Gramm
阅读更多...
Java字符串操作技巧之语法、示例与应用场景分析

Java字符串操作技巧之语法、示例与应用场景分析

《Java字符串操作技巧之语法、示例与应用场景分析》在Java算法题和日常开发中,字符串处理是必备的核心技能,本文全面梳理Java中字符串的常用操作语法,结合代码示例、应用场景和避坑指南,可快速掌握字... 目录引言1. 基础操作1.1 创建字符串1.2 获取长度1.3 访问字符2. 字符串处理2.1 子字
阅读更多...
Python基础语法中defaultdict的使用小结

Python基础语法中defaultdict的使用小结

《Python基础语法中defaultdict的使用小结》Python的defaultdict是collections模块中提供的一种特殊的字典类型,它与普通的字典（dict）有着相似的功能,本文主要... 目录示例1示例2python的defaultdict是collections模块中提供的一种特殊的字
阅读更多...
Spring Boot3虚拟线程的使用步骤详解

Spring Boot3虚拟线程的使用步骤详解

《SpringBoot3虚拟线程的使用步骤详解》虚拟线程是Java19中引入的一个新特性,旨在通过简化线程管理来提升应用程序的并发性能,：本文主要介绍SpringBoot3虚拟线程的使用步骤,... 目录问题根源分析解决方案验证验证实验实验1：未启用keep-alive实验2：启用keep-alive扩展建
阅读更多...
C++ 中的 if-constexpr语法和作用

C++ 中的 if-constexpr语法和作用

《C++中的if-constexpr语法和作用》if-constexpr语法是C++17引入的新语法特性,也被称为常量if表达式或静态if（staticif）,：本文主要介绍C++中的if-c... 目录1 if-constexpr 语法1.1 基本语法1.2 扩展说明1.2.1 条件表达式1.2.2 fa
阅读更多...
Java终止正在运行的线程的三种方法

Java终止正在运行的线程的三种方法

《Java终止正在运行的线程的三种方法》停止一个线程意味着在任务处理完任务之前停掉正在做的操作,也就是放弃当前的操作,停止一个线程可以用Thread.stop()方法,但最好不要用它,本文给大家介绍了... 目录前言1. 停止不了的线程2. 判断线程是否停止状态3. 能停止的线程–异常法4. 在沉睡中停止5
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2