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。

相关文章

MyBatis常用XML语法详解

MyBatis常用XML语法详解

《MyBatis常用XML语法详解》文章介绍了MyBatis常用XML语法,包括结果映射、查询语句、插入语句、更新语句、删除语句、动态SQL标签以及ehcache.xml文件的使用,感兴趣的朋友跟随小... 目录1、定义结果映射2、查询语句3、插入语句4、更新语句5、删除语句6、动态 SQL 标签7、ehc
阅读更多...
JDK21对虚拟线程的几种用法实践指南

JDK21对虚拟线程的几种用法实践指南

《JDK21对虚拟线程的几种用法实践指南》虚拟线程是Java中的一种轻量级线程,由JVM管理,特别适合于I/O密集型任务,：本文主要介绍JDK21对虚拟线程的几种用法,文中通过代码介绍的非常详细,... 目录一、参考官方文档二、什么是虚拟线程三、几种用法1、Thread.ofVirtual().start(
阅读更多...
Java 虚拟线程的创建与使用深度解析

Java 虚拟线程的创建与使用深度解析

《Java虚拟线程的创建与使用深度解析》虚拟线程是Java19中以预览特性形式引入,Java21起正式发布的轻量级线程,本文给大家介绍Java虚拟线程的创建与使用,感兴趣的朋友一起看看吧... 目录一、虚拟线程简介1.1 什么是虚拟线程？1.2 为什么需要虚拟线程？二、虚拟线程与平台线程对比代码对比示例：三
阅读更多...
Java 线程池+分布式实现代码

Java 线程池+分布式实现代码

《Java线程池+分布式实现代码》在Java开发中,池通过预先创建并管理一定数量的资源,避免频繁创建和销毁资源带来的性能开销,从而提高系统效率,：本文主要介绍Java线程池+分布式实现代码,需要... 目录1. 线程池1.1 自定义线程池实现1.1.1 线程池核心1.1.2 代码示例1.2 总结流程2. J
阅读更多...
Java JUC并发集合详解之线程安全容器完全攻略

Java JUC并发集合详解之线程安全容器完全攻略

《JavaJUC并发集合详解之线程安全容器完全攻略》Java通过java.util.concurrent（JUC）包提供了一整套线程安全的并发容器,它们不仅是简单的同步包装,更是基于精妙并发算法构建... 目录一、为什么需要JUC并发集合？二、核心并发集合分类与详解三、选型指南：如何选择合适的并发容器？在多
阅读更多...
Java中如何正确的停掉线程

Java中如何正确的停掉线程

《Java中如何正确的停掉线程》Java通过interrupt()通知线程停止而非强制,确保线程自主处理中断,避免数据损坏,线程池的shutdown()等待任务完成,shutdownNow()强制中断... 目录为什么不强制停止为什么 Java 不提供强制停止线程的能力呢？如何用interrupt停止线程s
阅读更多...
python 线程池顺序执行的方法实现

python 线程池顺序执行的方法实现

《python线程池顺序执行的方法实现》在Python中,线程池默认是并发执行任务的,但若需要实现任务的顺序执行,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋... 目录方案一：强制单线程（伪顺序执行）方案二：按提交顺序获取结果方案三：任务间依赖控制方案四：队列顺序消
阅读更多...
Python 函数详解:从基础语法到高级使用技巧

Python 函数详解:从基础语法到高级使用技巧

《Python函数详解:从基础语法到高级使用技巧》本文基于实例代码,全面讲解Python函数的定义、参数传递、变量作用域及类型标注等知识点,帮助初学者快速掌握函数的使用技巧,感兴趣的朋友跟随小编一起... 目录一、函数的基本概念与作用二、函数的定义与调用1. 无参函数2. 带参函数3. 带返回值的函数4.
阅读更多...
SpringBoot实现虚拟线程的方案

SpringBoot实现虚拟线程的方案

《SpringBoot实现虚拟线程的方案》Java19引入虚拟线程,本文就来介绍一下SpringBoot实现虚拟线程的方案,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,... 目录什么是虚拟线程虚拟线程和普通线程的区别SpringBoot使用虚拟线程配置@Async性能对比H
阅读更多...
在Java中实现线程之间的数据共享的几种方式总结

在Java中实现线程之间的数据共享的几种方式总结

《在Java中实现线程之间的数据共享的几种方式总结》在Java中实现线程间数据共享是并发编程的核心需求,但需要谨慎处理同步问题以避免竞态条件,本文通过代码示例给大家介绍了几种主要实现方式及其最佳实践,... 目录1. 共享变量与同步机制2. 轻量级通信机制3. 线程安全容器4. 线程局部变量（ThreadL
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2