快学Scala第20章----Actor

16-07-14 来源：[db:作者]

收藏我要投稿

本章要点

每个actor都要扩展Actor类并提供act方法要往actor发送消息，可以用actor ! message . 消息发送是异步的：“发完就忘” 要接收消息，actor可以调用receive或react，通常是在循环中这样做。 receive/react的参数是由case语句组成的代码块(从技术上讲，这是一个偏函数) 不同actor之间不应该共享状态。总是使用消息来发送状态不要直接调用actor的方法。通过消息进行通信。避免同步消息—也就是说将发送消息和等待响应分开不同actor可以通过react而不是receive来共享线程，前提是消息处理器的控制流转足够简单让actor挂掉是OK的，前提是你有其他actor监控着actor的生死。用链接来设置监控关系。

创建和启动Actor

actor是扩展自Actor特质的类。该特质带有一个抽象方法act。我们可以重写这个方法来指定该actor的行为。

import scala.actors.Actor

class HiActor extends Actor {
  def act() {
    while(true) {
      receive {
        case "Hi" => println("Hello")
      }
    }
  }
}

act方法和Java中Runnable接口中的run方法很相似，而且不同actor的act方法也是并行运行的。不同的是， actor针对响应消息做了优化，而线程可以开展任意的活动。
要启动一个actor，我们需要构造一个实例，并调用start方法：

val actor1 = new HiActor
actor1.start()

有时，我们需要临时创建actor而不是定义一个类。Actor伴生对象有一个actor方法来创建和启动actor：

import scala.actors.Actor._

val actor2 = actor {
  while(true) {
    receive {
      case "Hi" => println("Hello")
    }
  }
}

说明： 有时，一个匿名的actor需要向另一个actor发送指向自己的引用。这个引用可以通过self属性获取。

发送消息

actor是一个处理异步消息的对象。消息可以是任何对象。要发送消息，我们可以使用为actor定义的!操作符：

actor1 ! "Hi"

消息被送往该actor，当前的线程继续执行。我们也可以阻塞当前线程，直到等到回复（这并不常见）。
一个好的做法是使用样例类作为消息。这样一来，actor就可以使用模式匹配来处理消息。例如，我们有一个检查信用卡诈骗的actor。我们可能会想要向他发消息说某个记账动作正在执行。

case class Charge(creditCardNumber: Long, merchant: String, amount: Double)

// 样例类的一个对象发送给actor
fraudControl ! Charge(4111111111111111L, "Fred's Bait and Tackle", 19.95)

// 假定该actor的act方法包含一个如下语句
receive {
  case Charge(ccnum, merchant, amt) => ...
}

接收消息

发送到actor的消息被存放在一个“邮箱”中。receive方法从邮箱获取下一条消息并将它传递给它的参数，该参数是一个偏函数。例如：

receive {
  case Deposit(amount) => ...
  case Withdraw(amount) => ...
}

receive的代码块被转换成一个类型为PartialFunction[Any, T]的对象，其中T是case语句=>操作符右边的表达式的计算结果的类型。这是个“偏”函数，因为它只对那些能够匹配其中一个case语句的参数有定义。
说明： 消息投递的过程是异步的。所以你不能让你的程序依赖任何特定顺序的消息。
如果在receive方法被调用时并没有消息，则该调用会阻塞，直到有消息抵达。
如果邮箱中没有任何消息可以被偏函数处理，则对receive方法的调用也会阻塞，直到一个可以匹配的消息抵达。
注意： 邮箱可能被那些不与任何case语句匹配的消息占满。你可以添加一个 case _来处理任意的消息。
actor运行在单个线程中，在actor作用于内的资源是不会发生争用的。例如：

class AccountActor extends Actor {
  private var balance = 0.0

  def act() {
    while(true) {
      receive {
        case Deposit(amount) => balance += amount
        case Withdraw(amount) => balance -= amount
        ...
      }
    }
  }
}

**注意： **actor可以安全的修改它自己的数据。但是如果它修改了在不同actor之间共享的数据，那么争用状况就可能出现。
不要在不同的actor中使用共享的对象，这样为了同步会牺牲掉效率。除非你知道对这个对象的访问是线程安全的。

向其他Actor发送消息

当运算被拆分到不同actor来并行处理问题的各个部分时，这些处理结果最终要被收集到一起。actor可以将结果存入到一个线程安全的数据结构当中，比如一个并发的哈希映射。但是使用共享的数据结构会导致资源争用，为了同步，势必有的actor会被阻塞，这样计算效率就会下降。actor鼓励将计算结果通过消息的方式发送到另一个actor(这挺像在Qt中的使用信号和槽样)。
一个actor是如何知道应该往哪里发送计算结果的呢？这里有几个设计选择：
1. 可以有一些全局的actor。不过，当actor数量很多时，这个方案的伸缩性不好。
2. actor可以构造成带有指向一个或更多actor的引用。
3. actor可以接收带有指向另一个actor的引用的消息。在请求中提供一个actor引用是很常见的做法。例如：

actor ! Computer(data, continuation)  // continuation是另一个actor，当结果被计算出来的时候应该调用该actor

actor可以返回消息给发送方。receive方法会把sender字段设为当前消息的发送方。

消息通道

除了在你的应用程序中，对actor共享引用的做法外，你还可以共享消息通道给它们。这样做的好处：
1. 消息通道是类型安全的—-你只能发送或接受某个特定类型的消息。
2. 你不会不小心通过消息通道调用到某个actor的方法。
消息通道可以是一个OutputChannel(带有!方法)，也可以是一个InputChannel(带有receive或react方法)。Channel类同时扩展OutputChannel和InputChannel特质。
要构造一个消息通道，你可以提供一个actor：

val channel = new Channel[Int](someActor)

如果你不提供构造参数，那么消息通道就会绑定到当前执行的这个actor上

case class Computer(input: Seq[Int], result: OutputChannel[Int])
class Computer extends Actor {
  public void act() {
    while(true) {
      receive {
        case Computer(input, out) => { val answer = ...; out ! answer }
      }
    }
  }
}


actor {
  val channel = new Channel[Int]
  val computeActor: Computer = ...
  val input: Seq[Double] = ...
  computeActor ! Compute(input, channel)
  channel.receive {
    case x => ...
  }
}

同步消息和Future

actor 可以发送一个消息并等待回复，用!?操作符即可。例如：

val reply = account !? Deposit(1000)
reply match {
  case Balance(bal) => println("Current Balance: " + bal)
}

上面的代码发送消息后就会被阻塞，直到接收到回复。因此接收方必须返回一个消息给发送方：

receive {
  case Deposit(amount) => {
    balance += amount; sender ! Balance(balance)
    ...
  }
}

说明： 除了用sender ! Balance(balance)，你也可以写reply(Balance(balance))。
注意： 同步消息容易引发死锁。通产而言，避免在actor的act方法里执行阻塞调用。
你也可以设置等待时间，超时后将会收到一个Actor.TIMEOUT对象。

actor {
  woker ! Task (data, self)
  receiveWithin(seconds * 1000) {
    case Result(data) => ...
    case TIMEOUT => log(...)
  }
}

除了等待对方返回结果之外，你也可以选择接收一个future—–这是一个将在结果可用时产出结果的对象。使用!!方法即可做到：

val replyFuture = account !! Deposit(1000)

isSet方法会检查结果是否已经可用。要接收该结果，使用函数调用的表达法：

val reply = replyFuture()

这个调用将会阻塞，直到回复被发送。

共享线程

如果程序中使用了大量的actor，而为每个actor创建一个单独的线程开销会很大。我们可以在同一个线程中运行多个actor，这样做的前提是每个消息处理函数只需要做比较小规模的工作，然后就继续等待下一条消息。
在Scala中，可以使用react方法接受一个偏函数，并将它添加到邮箱，然后退出。嘉定我们有两个嵌套的react语句：

react {  // 偏函数f1
  case Withdraw(amount) =>
  react {  // 偏函数f2
    case Confirm() =>
      println("Confirming " + amount)
  }
}

在这里，第一个react的调用将f1与actor的邮箱关联起来，然后退出。当Withdraw消息抵达时，f1被调用。偏函数f1也调用了react。这次调用把f2与actor的邮箱关联起来，然后退出。当Confirm消息到达时，f2被调用。
由于react会退出，因此你不能简单的将它放在while循环中。例如：

def act() {
  while(true) {
    react {
      case Withdraw(amount) => println("Withdrawing " + amount)
    }
  }
}

在这里，当act被调用时，对react的调用将f1与邮箱关联起来，然后退出。当f1被调用时，它将会处理这条消息。不过，f1没有任何办法返回到循环当中—–它只不过是一个小小的函数:

{ case Withdraw(amount) => println("Withdrawing " + amount) }

解决办法之一是在消息处理器中再次调用act方法：

def act() {
  react {
    case Withdraw(amount) => {
      println("Withdrawing " + amount)
      act()
    }
  }
}

这样做意味着用一个无穷递归替换掉无穷循环。这个递归不会占用很大的栈空间。每次对react的调用都会抛出异常，从而清栈。
有一些“控制流转组合子”可以自动产出这些循环。loop组合子可以制作一个无穷循环：

def act() {
  loop {
    react {
      case Withdraw(amount) => println("Withdrawing " + amount)
    }
  }
}

也可以使用循环条件:

loopWhile( count < max ) {
  react {
    ...
  }

}

Actor的生命周期

actor的act方法在actor的start方法被调用时开始执行。actor的如下情形之一会终止执行：
1. act 方法返回
2. act方法由于异常被终止
3. actor调用exit方法。
**说明： **exit方法是个受保护的方法，它只能被Actor的子类中调用。例如：

val actor1 = actor {
  while (true) {
    receive {
      case "Hi" => println("Hello")
      case "Bye" => exit()
    }
  }
}

其他方法不能调用exit来终止一个actor。
当actor因一个异常终止时，退出的原因就是UncaughtException样例类的一个实例。
- actor : 抛出异常的actor。
- message: Some(msg), 其中msg是该actor处理的最后一条消息；或者None，如果actor在没来得及处理任何消息之前就挂掉的话。
- sender: Some(channel)，其中channel是代表最后一条消息的发送方的输出消息通道；或者None，如果actor在没来得及处理任何消息之前就挂掉的话。
- thread： actor退出时所在的线程
- cause：相应的异常。

将多个Actor链接在一起

如果你将两个actor链接在一起，则每以个都会在另一个终止执行的时候得到通知。要建立这个关系，只需要简单的以另一个actor的引用调用调用link方法：

def act() {
  link(master)
  ...
}

链接是双向的。如果监管actor在数个工作actor中分发工作任务，当某个工作actor挂掉的时候，监管actor应该知道，以便相应的工作可以被重新指派。反过来，如果监管actor挂掉了，工作actor也应该知道，以便可以停止工作(可以有多个actor实现HA，参考Zookeeper)。
注意： 尽管链接是双向的，但link方法并不对称。你不能将link(worker)替换成work.link(self)。该方法必须由请求链接的actor调用。
默认情况下，只要当前actor链接到的actor中有一个以非‘normal原因退出，当前actor就会终止。在这种情况下，退出原因和链接到的那个actor的退出的原因相同。
actor可以改变这个行为，做法是设置trapExit为true。这样修改后，actor会接收到一个类型为Exit的消息，该消息包含了那个正在终止的actor和退出的原因。

override def act() {
  trapExit = true
  link(worker)
  while(true) {
    receive {
      ...
      case Exit(linked, UncaughtException(_,_,_,_, cause)) => ...
      case Exit(linked, reason) => ...
    }
  }
}

在操作actor时，允许它们挂掉是很正常的。只需要把每个actor链接到一个监管actor，由它来处理失败的actor，例如，将它们的工作重新分派，或重新的启动它们。

点击复制链接与好友分享!回本站首页