Scala Actor

二、什么是Scala Actor

1．概念
Scala中的Actor能够实现并行编程的强大功能，它是基于事件模型的并发机制，Scala是运用消息（message）的发送、接收来实现多线程的。使用Scala能够更容易地实现多线程应用的开发。
2．传统java并发编程与Scala Actor编程的区别

对于Java，我们都知道它的多线程实现需要对共享资源（变量、对象等）使用synchronized 关键字进行代码块同步、对象锁互斥等等。而且，常常一大块的try…catch语句块中加上wait方法、notify方法、notifyAll方法是让人很头疼的。原因就在于Java中多数使用的是可变状态的对象资源，对这些资源进行共享来实现多线程编程的话，控制好资源竞争与防止对象状态被意外修改是非常重要的，而对象状态的不变性也是较难以保证的。而在Scala中，我们可以通过复制不可变状态的资源（即对象，Scala中一切都是对象，连函数、方法也是）的一个副本，再基于Actor的消息发送、接收机制进行并行编程
3． Actor方法执行顺序
1.首先调用start()方法启动Actor
2.调用start()方法后其act()方法会被执行
Scala Actor向Actor发送消息
4．发送消息的方式
! 发送异步消息，没有返回值。
!? 发送同步消息，等待返回值。
!! 发送异步消息，返回值是 Future[Any]。

三、 Actor实战

如果发现Actor的包引入不了，解决办法是：
选中项目右键—>Open Module Settings—>Library–>添加jar(scala-2.10.6.zip中的\lib*.jar)
上面的右侧就是我们引入的jar包

1．第一个例子

package cn.toto.scala.day2
import scala.actors.Actor//这种执行完成之后就停止了。
object MyActor1 extends Actor {//重新act方法def act(): Unit = {for(i <- 1 to 10) {println("actor-1 " + i)Thread.sleep(3000)}}
}object MyActor2 extends Actor {def act(): Unit = {for(i <- 1 to 10) {println("actor-2 " + i)Thread.sleep(2500)}}
}object ActorTest extends App {//因为是object类型的，不需要new,可以直接调用start方法MyActor1.start()MyActor2.start()
}

说明：上面分别调用了两个单例对象的start()方法，他们的act()方法会被执行，相同与在java中开启了两个线程，线程的run()方法会被执行
注意：这两个Actor是并行执行的，act()方法中的for循环执行完成后actor程序就退出了

2．第二个例子（可以不断地接收消息）
receive相当于是创建线程和销毁线程的过程。

package cn.toto.scala.day2import scala.actors.Actor/*** Created by toto on 2017/7/1.*/
class MyActor extends Actor {override def act(): Unit = {while(true) {receive {case "start" => {println("starting...")Thread.sleep(5000)println("stared")}case "stop" => {println("stoping ...")Thread.sleep(5000)println("stopped ...")}}}}
}//伴生对象，说明在一个Actor内部是串行的，如果想是并发执行的，可以创建多个actor
object MyActor {def main(args: Array[String]): Unit = {val actor = new MyActoractor.start()//发送异步消息，感叹号就相当于是一个方法，可以在感叹号后面的字符串两边加载上括号actor ! "start"actor ! "stop"println("xiao ")}
}

说明：在act()方法中加入了while (true) 循环，就可以不停的接收消息
注意：发送start消息和stop的消息是异步的，但是Actor接收到消息执行的过程是同步的按顺序执行

3．第三个例子（react方式会复用线程，比receive更高效）
react类似线程池机制，可以复用线程。

package cn.toto.scala.day2import scala.actors.Actor/*** Created by toto on 2017/7/1.*/
class YourActor extends Actor {override def act(): Unit = {loop {//这里是一个偏函数react {case "start" => {println("starting...")Thread.sleep(5000)println("started")}case "stop" => {println("stopping...")Thread.sleep(8000)println("stopping")}}}}
}/*** 运行结果：* 消息发送完成!* starting...* started* stopping...* stopping*/
object YourActor {def main(args: Array[String]): Unit = {val actor = new YourActoractor.start()actor ! "start"actor ! "stop"println("消息发送完成!")}
}运行结果是：
消息发送完成!
starting...
started
stopping...
stopping...

说明： react 如果要反复执行消息处理，react外层要用loop，不能用while

4．第四个例子（结合case class发送消息）

package cn.toto.scala.day2import scala.actors.Actorcase class SyncMsg(id:Int,msg:String)
case class AsyncMsg(id:Int,msg:String)
case class ReplyMsg(id:Int,msg:String)/*** Created by toto on 2017/7/1.*/
class AppleActor extends Actor {def act(): Unit = {while (true) {receive {case "start" => println("starting ...")case SyncMsg(id, msg) => {println(id + ",sync " + msg)Thread.sleep(5000)//这里又开了一个子线程，这个在下面的方法中执行//sender也是一个方法。看可以通过消息的发送者发送sender ! ReplyMsg(3,"finished")}case AsyncMsg(id, msg) => {println(id + ",async " + msg)Thread.sleep(5000)}}}}
}object AppleActor {/*** 总体的运行结果是：* 异步消息发送完成* 1,async hello actor* None* false* 123* 2,sync hello actor* 2,sync hello actor* true* ReplyMsg(3,finished)* @param args*/def main(args: Array[String]): Unit = {val a = new AppleActora.start()//异步消息a ! AsyncMsg(1,"hello actor")println("异步消息发送完成")//同步消息(因为这里的最长时间是1秒，但是SyncMsg中休眠的时间更长，为5秒，所以这里是None)//同步消息，表示需要等待val content = a.!? (1000,SyncMsg(2,"hello actor"))println(content)//异步但是返回结果，这个结果会被放到future中。然后进行返回。val reply = a !! SyncMsg(2,"hello actor")//replay是否已经完成通过isSet类判断println(reply.isSet);println("123")//通过apply方法拿到里面的值，这个apply是feture中的方法。val c = reply.apply()//拿到之后，reply中的值相当于被设置了。所以c为trueprintln(reply.isSet)println(c)}
}

Future的使用

package cn.toto.scala.day2;import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;import java.util.concurrent.*;/*** Created by ZhaoXing on 2016/6/27.* Feture相当于给出了一种期望，就是说可能立即不会有结果，但是执行完毕或过一段时间后会有结果*/
public class FutureDemo {/*** 运行结果：* main false* 正在查找数据* pool-1-thread-1 数据查找完毕* true* 9527* @param args* @throws Exception*/public static void main(String[] args) throws Exception{ExecutorService threadPool = Executors.newCachedThreadPool();Future<Integer> future = threadPool.submit(new Callable<Integer>() {@Overridepublic Integer call() throws Exception {System.out.println("正在查找数据");Thread.sleep(4000);System.out.println(Thread.currentThread().getName() + " 数据查找完毕");return 9527;}});System.out.println(Thread.currentThread().getName() + " "  + future.isDone());//System.out.println(future.get(1000, TimeUnit.MILLISECONDS));Thread.sleep(6000);System.out.println(future.isDone());System.out.println(future.get());}
}

使用Actor进行wordCount

package cn.toto.scala.day2import scala.actors.{Actor, Future}
import scala.collection.mutable.ListBuffer
import scala.collection.parallel.Task
import scala.io.Source/*** Created by toto on 2017/7/1.** 使用线程的方式计算单词出现的频率*/
case class SbTask(fn:String)
case object StopTaskclass WordCountByActorTask extends Actor {override def act(): Unit = {while(true) {receive {case SbTask(f) => {//通过Source.fromFile(f)读取文件，获取每行并且把它转换成Listval lines : List[String] = Source.fromFile(f).getLines().toList;//将lines list里面的内容合并，然后并且将它split.val words : List[String] = lines.flatMap(_.split(" "))//将word这些单词变成map的并且是元组类型的，每个都是1，----,记着对他们进行过分组，接着就是计算单词val result: Map[String, Int]= words.map((_, 1)).groupBy(_._1).mapValues(_.size)sender ! result}case StopTask => {exit()}}}}
}object WordCountByActor {def main(args: Array[String]): Unit = {val replys = new ListBuffer[Future[Any]]val results = new ListBuffer[Map[String,Int]]val files = Array("E://wordcount//input//a.txt","E://wordcount//input//b.txt","E://wordcount//input//c.txt")//每个文件就启动多少个actor,并将每个actor处理后得到的结果存储到List中for(f <- files) {val t = new WordCountByActorTaskt.start()//通过SbTask的方式发送文件名val reply:Future[Any] = t !! SbTask(f)//将处理到的结果放到ListBuffer中。replys += reply}//对获得的replay的值进行计算。while (replys.size > 0) {val dones: ListBuffer[Future[Any]] = replys.filter(_.isSet)for(f <- dones) {//从relay中拿值，并将它转成mapresults += f.apply().asInstanceOf[Map[String, Int]]//计算完成之后，将这个replay移除掉。replys -= f}Thread.sleep(500)}println(results.flatten.groupBy(_._1).mapValues(_.foldLeft(0)(_+_._2)))}
}

Scala Actor,receive不断接收消息，react复用线程，结合case class的actor,Future使用,使用Actor进行wordCount相关推荐

Akka 接收消息超时的处理_Receive Timeout
2019独角兽企业重金招聘Python工程师标准>>> Akka 接收信息超时的处理_Receive Timeout The UntypedActorContext setRecei ...
ActiveMQ中消费者是如何接收消息的（二）
上篇文章大致讲述了同步消费者和异步消费者接收消息的异同(详见<ActiveMQ中消费者是如何接收消息的(一)>http://manzhizhen.iteye.com/blog/209413 ...
ActiveMQ 部署及发送接收消息
ActiveMQ 部署及发送接收消息一. 下载下载地址:http://activemq.apache.org/ 我这里使用的版本为当前最新5.8.0. 下载版本有Windows ...
【Spring】使用Spring和AMQP发送接收消息（下）
为什么80%的码农都做不了架构师?>>> 上篇讲了RabbitMQ连接工厂的作用是用来创建RabbitMQ的连接,本篇就来讲讲RabbitMQ的发送消息.通过RabbitMQ发 ...
java mqtt客户端_java 实现mqtt发送和接收消息客户端具体用法及测试代码
注:客户端代码实现请看我的上一篇 1mqtt发送消息发送时不用多讲,每次发送肯定需要运行一次发送消息方法 MyMqttClient mqttClient = new MyMqttClient(); ...
c#UDP发送接收消息
UDP作为一种简单的.面向数据报的无连接的协议,虽然提供的是不可靠的服务,但是从速度上.一对多传输方面比TCP有着很大的优势.本文主要讲解UDP信息的发送和接收. Demo界面图如下: 首先打开程序e ...
Akka和Actor一起工作的消息《ten》译
Dependency 在项目中添加以下依赖项: Introduction 在前面的主题解说中,我们解释了如何构建大型的Actor系统,即如何表示组件,如何在层次结构中排列Actor.在这一部分中,我们 ...
微信公众号三方平台开发【代微信公众号接收消息事件并响应】
今天要讲的是代微信公众号接收消息并进行回复,这里要用到的就是咱们之前在微信公众号三方平台开发[帐号注册.平台创建]里写到的"公众号消息与事件接收URL",在接收的目标方法里,我们首 ...
Unity丛林战争学习01_TCP_异步接受连接和异步接收消息
1.服务器端服务器端所要做的事: ①创建服务器端套接字(Socket) ②绑定套接字(Bind) ③监听连接(Listen) ④异步接受连接(BeginAccept) ⑤结束异步接受连接,返回客户端 ...

Scala Actor,receive不断接收消息，react复用线程，结合case class的actor,Future使用,使用Actor进行wordCount

二、什么是Scala Actor

三、 Actor实战

Scala Actor,receive不断接收消息，react复用线程，结合case class的actor,Future使用,使用Actor进行wordCount相关推荐

最新文章

热门文章

Scala Actor,receive不断接收消息，react复用线程，结合case class的actor,Future使用,使用Actor进行wordCount

二、 什么是Scala Actor

三、 Actor实战

Scala Actor,receive不断接收消息，react复用线程，结合case class的actor,Future使用,使用Actor进行wordCount相关推荐

最新文章

热门文章

二、什么是Scala Actor