Akka并发编程——第七节:Actor模型(六)

主要内容:
1. Typed Actor定义
2. Typed Actor创建
3. 消息发送

1. Typed Actor定义

Akka中的Typed Actor是Active Objects设计模式的实现,Active Objects模式将方法的执行和方法的调用进行解耦合,从而为程序引入并发性。Typed Actor由公用的接口和对应实现两部分构成,其后面深层次的实现使用的是代理模式,即通过使用JDK中的动态代理来实现,在调用接口的方法时自动分发到实现接口的对象上。Typed Actor的定义[ ]如下所示。

trait Squarer {
    //fire-and-forget消息
    def squareDontCare(i: Int): Unit
    //非阻塞send-request-reply消息
    def square(i: Int): Future[Int]
    //阻塞式的send-request-reply消息
    def squareNowPlease(i: Int): Option[Int]
    //阻塞式的send-request-reply消息
    def squareNow(i: Int): Int
  }

  class SquarerImpl(val name: String) extends Squarer {
    def this() = this("SquarerImpl")

    def squareDontCare(i: Int): Unit = i * i
    def square(i: Int): Future[Int] = Promise.successful(i * i).future
    def squareNowPlease(i: Int): Option[Int] = Some(i * i)
    def squareNow(i: Int): Int = i * i
  }

trait Squarer中定义了4个方法:
(1)def squareDontCare(i: Int): Unit方法:返回值类型为Unit,它类似于Untyped Actor中的fire-and-forget消息发送模型,即!和tell方法调用。
(2)def square(i: Int): Future[Int]:返回值类型为Future[Int],它类似于Untyped Actor中的send-request-reply消息发送模型,即?和ask方法调用,此种调用是非阻塞的。
(3)def squareNowPlease(i: Int): Option[Int]:返回值类型为Option[Int](Option类可以是scala.Option[_]也可以是akka.japi.Option

2. 创建Typed Actor

通过下列代码创建Typed Actor实例。

//直接通过默认的构造函数创建Typed Actor
val mySquarer: Squarer =TypedActor(system).typedActorOf(TypedProps[SquarerImpl]())
//直接通过默认的构造函数创建Typed Actor并指定Typed Actor名称
val mySquarer: Squarer =TypedActor(system).typedActorOf(TypedProps[SquarerImpl](),"mySquarer")
//通过非默认的构造函数创建Typed Actor并指定Typed Actor名称
val otherSquarer: Squarer = TypedActor(system).typedActorOf(TypedProps(classOf[Squarer],new SquarerImpl("SquarerImpl")), "otherSquarer")

上面代码演示的是使用构造函数和非默认构造函数创建Typed Actor,其中Squarer为代理的类型,SquarerImpl为具体实现的类型。

3. 消息发送

//fire-forget消息发送
  mySquarer.squareDontCare(10)

  //send-request-reply消息发送
  val oSquare = mySquarer.squareNowPlease(10)

  val iSquare = mySquarer.squareNow(10)

  //Request-reply-with-future 消息发送
  val fSquare = mySquarer.square(10)
  val result = Await.result(fSquare, 5 second)

代码mySquarer.squareDontCare(10)是单向消息发送,方法将在另外一个线程上异步地执行;val oSquare = mySquarer.squareNowPlease(10)、val iSquare = mySquarer.squareNow(10)为Request-reply消息发送,在特定时间内以阻塞的方式执行,对于.squareNowPlease(10)方法如果在对应时间内没有返回结果则返回值为None,否则返回值为Option[Int]类型,对于squareNow(10)方法如果在对应时间内无返回值则会抛出异常java.util.concurrent.TimeoutException,否则返回Int类型值;val fSquare = mySquarer.square(10)为Request-reply-with-future式的消息发送,以非阻塞的方式执行,可以通过val result = Await.result(fSquare, 5 second)获取执行结果。完整代码如下所示。

/*
 * Typed Actor
 */
object Example_01 extends  App {

  import akka.event.Logging
  import scala.concurrent.{ Promise, Future }
  import akka.actor.{ TypedActor, TypedProps }
  import scala.concurrent.duration._

  trait Squarer {
    //fire-and-forget消息
    def squareDontCare(i: Int): Unit
    //非阻塞send-request-reply消息
    def square(i: Int): Future[Int]
    //阻塞式的send-request-reply消息
    def squareNowPlease(i: Int): Option[Int]
    //阻塞式的send-request-reply消息
    def squareNow(i: Int): Int
  }

  class SquarerImpl(val name: String) extends Squarer {
    def this() = this("SquarerImpl")

    def squareDontCare(i: Int): Unit = i * i
    def square(i: Int): Future[Int] = Promise.successful(i * i).future
    def squareNowPlease(i: Int): Option[Int] = Some(i * i)
    def squareNow(i: Int): Int = i * i
  }

  val system = ActorSystem("TypedActorSystem")
  val log = Logging(system, this.getClass)

  //使用默认构造函数创建Typed Actor
  val mySquarer: Squarer =
    TypedActor(system).typedActorOf(TypedProps[SquarerImpl](),"mySquarer")

  //使用非默认构造函数创建Typed Actor
    val otherSquarer: Squarer =
      TypedActor(system).typedActorOf(TypedProps(classOf[Squarer],
        new SquarerImpl("SquarerImpl")), "otherSquarer")


  //fire-forget消息发送
  mySquarer.squareDontCare(10)

  //send-request-reply消息发送
  val oSquare = mySquarer.squareNowPlease(10)

  log.info("oSquare="+oSquare)

  val iSquare = mySquarer.squareNow(10)
  log.info("iSquare="+iSquare)

  //Request-reply-with-future 消息发送
  val fSquare = mySquarer.square(10)
  val result = Await.result(fSquare, 5 second)

  log.info("fSquare="+result)

  system.shutdown()
}

代码运行结果如下:
[INFO] [03/21/2016 21:15:50.592] [main] [Example12_9(akka://TypedActorSystem)]oSquare=Some(100)[INFO][03/21/201621:15:50.649][main][Example129(akka://TypedActorSystem)] iSquare=100
[INFO] [03/21/2016 21:15:50.649] [main] [Example12_9$(akka://TypedActorSystem)] fSquare=100

本页内容版权归属为原作者,如有侵犯您的权益,请通知我们删除。

KafKa(0.10)安装部署和测试 - 2016-06-26 17:06:48

Step 1: 下载代码 http://kafka.apache.org/downloads.html 0.10.0.0是最新版本。 当前的稳定版本是0.10.0.0。 下载 0.10.0.0 版本并且解压它。 tar -zxvf kafka_2.10-0.10.0.0.tgz  cd kafka_2.11-0.10.0.0 Step 2: 启动服务 运行kafka需要使用Zookeeper,所有你需要先启动一个Zookeeper服务器,如果你没有Zookeeper,你可以使用kafka自带打包和配置好的
spark for python developer 一书,说实在的,质量一般,但勉强可以作为python 工程师的入门资料,至此,这一时段的地铁译结束了,开始新的阅读旅程…… 对于 Python 的图形绘制和可视化, 有大量的工具和库,和我们最相关并且有趣的是:
 • Matplotlib 是Python 绘图库的鼻祖. Matplotlib 最初7由 John Hunter 创作, 他是开源软件的支持者,建立的 Matplotlib 是学术界和数据科学界最流行的绘图库之一。 Matplotlib 支持
前几天写“ Openstack liberty 创建实例快照源码分析1”时,在文章末尾留了一个悬念:在mitaka版本中优化了以ceph rbd做后端存储的云主机快照实现方式,本文将结合源码对这一改进具体分析。 首先来简单回顾下,liberty中’从镜像启动的云主机的快照原理’: 在liberty中,不管是在线快照还是离线快照,都需要先在本地生成临时快照文件,然后再上传到glance,最后以镜像模板的形式存在,用户可以使用该镜像模板启动新的云主机。下面一起来看看mitaka中云主机快照实现的过程: 与li
问题一:重启机器出现组件没法互通   openstack Failed to connect to server code:1006   重启后出现组件无法互通,很可能是因为防火墙问题。 简单的方法:iptables-F    注:不过,还是建议学习iptables,添加相应的规则 问题二:实时查看某些日志的输出: 例如查看实时日志nova-compute.log,以便实时调试组件问题 cd /var/log/nova/tail-f nova-compute.log 问题三:关闭selinux 修改 /e
【前言】 Docker是Docker.Inc公司开源的一个基于轻量级虚拟化技术的容器引擎项目,整个项目基于Go语言开发,并遵从Apache 2.0协议。通过分层镜像标准化和内核虚拟化技术,Docker使得应用开发者和运维工程师可以以统一的方式跨平台发布应用,并且以几乎没有额外开销的情况下提供资源隔离的应用运行环境。由于众多新颖的特性以及项目本身的开放性,Docker在不到两年的时间里迅速获得诸多IT厂商的参与,其中更是包括Google、Microsoft、VMware等业界行业领导者。同时,Docker在

Spark RDD使用详解1--RDD原理 - 2016-06-24 17:06:12

RDD简介       在集群背后,有一个非常重要的分布式数据架构,即弹性分布式数据集(Resilient Distributed Dataset,RDD)。RDD是Spark的最基本抽象,是对分布式内存的抽象使用,实现了以操作本地集合的方式来操作分布式数据集的抽象实现。RDD是Spark最核心的东西,它表示已被分区,不可变的并能够被并行操作的数据集合,不同的数据集格式对应不同的RDD实现。RDD必须是可序列化的。RDD可以cache到内存中,每次对RDD数据集的操作之后的结果,都可以存放到内存中,下一个
(上图为EasyStack创始人陈喜伦) 在过去十年里,IT行业一直在酝酿一场大变革,这就是云计算。在过去五年里,企业级IT在酝酿一场大变革,这就是以OpenStack为代表的开源IT。在过去的两年中,一批OpenStack创业公司在中国市场酝酿着企业IT生态的大变革,这就是以OpenStack为核心的企业云操作系统。 进入2016年以来,包括联想、神州数码、思科、新华三等老牌IT厂商以及IT生态链领导企业纷纷喊出新IT新生态的方向,要打造以企业云为基础的新企业级IT生态链。基于开源OpenStack的企
概述 本文要说的 TF-IDF 分布式实现,运用了很多之前 MapReduce 的核心知识点。算是 MapReduce 的一个小应用吧。 版权说明 著作权归作者所有。 商业转载请联系作者获得授权,非商业转载请注明出处。 本文作者: Q-WHai 发表日期: 2016年6月24日 本文链接: http://blog.csdn.net/lemon_tree12138/article/details/51747801 来源:CSDN 更多内容: 分类 大数据之 Hadoop 学前导读 本文并不打算再啰里啰嗦地讲
一  基本安装环境: 硬件资源 :服务器CPU需要支持虚拟化技术且在安装前需在BIOS设置中开启虚拟化功能,4网卡。 操作系统: debian7.4.0 在开始云平台的搭建工作前,还有一些服务器的基本配置要完成,包括网络的部署、系统的升级以及源地址的修改等等。 1.1  网络部署: 主机名 Eth0 Eth1 Eth2 (主控制节点)R320-1 10.9.32.1 10.0.0.10 10.8.8.1 (从控制节点)R320-2 10.9.32.2 10.0.0.20 10.8.8.2 (计算节点一)
即使不考虑数据节点出错后的故障处理,文件写入也是HDFS中最复杂的流程。本章以创建一个新文件并向文件中写入数据,然后关闭文件为例,分析客户端写文件时系统各节点的配合,如下图所示。 客户端调用DistributedFileSystem的create()方法创建文件,上图的步骤1,这时,DistributedFileSystem创建DFSOutputStream,并由远程过程调用,让名字节点执行同名方法,在文件系统的命名空间中创建一个新文件。名字节点创建新文件时,需要执行各种各样的检查,如名字节点处于正常工作