美文网首页
Actor 编程

Actor 编程

作者: Java旅行者 | 来源:发表于2017-11-14 21:45 被阅读0次

注:我们现在学的Scala Actor是scala 2.10.x版本及以前版本的Actor。
Scala在2.11.x版本中将Akka加入其中,作为其默认的Actor,老版本的Actor已经废弃

一、什么是 Actor

1. 概念

Scala中的Actor能够实现并行编程的强大功能,它是基于事件模型的并发机制,Scala是运用消息(message)的发送、接收来实现多线程的。

2. actor 方法执行顺序

  • 首先调用start()方法启动Actor
  • 调用start()方法后其act()方法会被执行
  • 向Actor发送消息

3. 发送消息的方式

方法 作用
! 发送异步消息,没有返回值。
!? 发送同步消息,等待返回值。
!! 发送异步消息,返回值是 Future[Any]。

二、 实战

1. 不断的接收消息

package cn.itcast.actor

import scala.actors.Actor

/**
  * Created by ZX on 2016/4/4.
  */
class MyActor extends Actor {

  override def act(): Unit = {
    while (true) {
      receive {
        case "start" => {
          println("starting ...")
          Thread.sleep(5000)
          println("started")
        }
        case "stop" => {
          println("stopping ...")
          Thread.sleep(5000)
          println("stopped ...")
        }
      }
    }
  }
}

object MyActor {
  def main(args: Array[String]) {
    val actor = new MyActor
    actor.start()
    actor ! "start"
    actor ! "stop"
    println("消息发送完成!")
  }
}

说明:在act()方法中加入了while (true) 循环,就可以不停的接收消息

注意:发送start消息和stop的消息是异步的,但是Actor接收到消息执行的过程是同步的按顺序执行

2. react方式会复用线程,比receive更高效

package cn.itcast.actor

import scala.actors.Actor

/**
  * Created by ZX on 2016/4/4.
  */
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("stopped ...")
        }
      }
    }
  }
}


object YourActor {
  def main(args: Array[String]) {
    val actor = new YourActor
    actor.start()
    actor ! "start"
    actor ! "stop"
    println("消息发送完成!")
  }
}

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

3. 结合case class发送消息

package cn.itcast.actor
import scala.actors.Actor

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 ! ReplyMsg(3,"finished")
        }
        case AsyncMsg(id, msg) => {
          println(id + ",async " + msg)
          Thread.sleep(5000)
        }
      }
    }
  }
}

object AppleActor {
  def main(args: Array[String]) {
    val a = new AppleActor
    a.start()
    //异步消息
    a ! AsyncMsg(1, "hello actor")
    println("异步消息发送完成")
    //同步消息
    //val content = a.!?(1000, SyncMsg(2, "hello actor"))
    //println(content)
    val reply = a !! SyncMsg(2, "hello actor")
    println(reply.isSet)
    //println("123")
    val c = reply.apply()
    println(reply.isSet)
    println(c)
  }
}
case class SyncMsg(id : Int, msg: String)
case class AsyncMsg(id : Int, msg: String)
case class ReplyMsg(id : Int, msg: String)

三、综合案例,Actor 实现 WordCount

package cn.itcast.scala.cn.itcast.scala.actor

import scala.actors.{Actor, Future}
import scala.collection.mutable.{HashSet, ListBuffer}
import scala.io.Source

object ActorWordCount {
  def main(args: Array[String]): Unit = {
    val replySet = new HashSet[Future[Any]]
    val resultSet = new ListBuffer[ResultTask]
    val taskList = new ListBuffer[WordCountTask]
    val files = List("F:/words1.txt","F:/words2.txt")
    for (f <- files) {
      val task = new WordCountTask
      val future = task.start() !! SubmitTask(f)
      replySet += future
      taskList += task
    }
    while (replySet.size > 0) {
      val toComplete = replySet.filter(_.isSet)
      for (f <- toComplete) {
        val result = f().asInstanceOf[ResultTask]
        resultSet += result
        replySet -= f
      }
    }
    val count = resultSet.flatMap(_.result).groupBy(_._1).mapValues(_.foldLeft(0)(_ + _._2))
    println(count)
    taskList.foreach(_ ! Stop)

  }
}

class WordCountTask extends Actor {
  override def act(): Unit = {
    loop {
      react {
        case SubmitTask(path) =>
          val result: Map[String, Int] = Source.fromFile(path).getLines().flatMap(_.split(" ")).map((_, 1)).toList.groupBy(_._1).mapValues(_.size)
          sender ! ResultTask(result)
        case Stop => exit()
      }
    }
  }
}

case class SubmitTask(path: String)
case class ResultTask(result: Map[String, Int])
case object Stop

相关文章

  • Actor 编程

    注:我们现在学的Scala Actor是scala 2.10.x版本及以前版本的Actor。Scala在2.11....

  • Scala编程:Actor入门

    Actor入门 引言 Scala的Actor类似于Java中的多线程编程。但是不同的是,Scala的Actor提供...

  • AKKA介绍(二)

    上一篇提到,Akka是Actor内单线程编程模型,不共享状态,使用消息在actor间通信。有效的提升actor内的...

  • Orleans(一) 整体介绍

    背景 Orleans是微软开源的Actor模型开发框架。 Actor模型此模型解决了并发编程时对资源竞争使用的问题...

  • Scala编程详解18:Actor入门

    Scala的Actor类似于Java中的多线程编程。但是不同的是,Scala的Actor提供的模型与多线程有所不同...

  • 漫谈并发编程:Actor模型

    0x00 前言 一般来说有两种策略用来在并发线程中进行通信:共享数据和消息传递。熟悉c和java并发编程的都会比较...

  • Actor并发编程模型浅析

    一.Actor模型介绍 在单核 CPU 发展已经达到一个瓶颈的今天,要增加硬件的速度更多的是增加 CPU 核的数目...

  • AKKA学习笔记

    有两种方式可以获取Actor引用,一是创建Actor,而是查找Actor。 创建Actor 一个actor系统通常...

  • 2017-07-08

    case 0: old actor close new actor() set new actor dxfr...

  • 剖析响应式编程的本质

    《基于Actor的响应式编程》计划分为三部分,第一部分剖析响应式编程的本质思想,为大家介绍何谓响应式编程(Reac...

网友评论

      本文标题:Actor 编程

      本文链接:https://www.haomeiwen.com/subject/uxlqvxtx.html