关于scala的Actor和经常配合使用的Future
Actor
Creating Actor & best practice
- 步骤一：实现trait
akka.actor.Actor即可创建一个Actor类型，其中的receive()方法，即处理接受信息的方法。
object Main {
def main(args: Array[String]):Unit = {
import akka.actor.Actor
class MyActor extends Actor {
override def receive: Unit = {
case "hello" => println("hello")
case _ => println("error mesg")
}
}
}
}
步骤二：通过
akka.actor.Props申明actor的构造函数步骤三：通过
akka.actor.ActorSystem或akka.actor.ActorContext创建actor：
import akka.actor._
class MyActor(param: Int) extends Actor {
override def receive: Receive = {
case _ => {
// 2. create with ActorContext within a Actor
context.actorOf(MyActor.props(param))
}
}
}
object MyActor {
// Best Practice:
// to declare what messages an Actor can receive in the companion object of the Actor
case class Message1(content:String)
case object Message2
// Best Practice:
// provide factory methods on the companion object of each Actor
// which help keeping the creation of suitable Props as close to the actor definition as possible
def props(param: Int): Props = {
Props(new MyActor(param))
}
def main(args: Array[String]): Unit = {
val props = Props(new MyActor(1))
// 1. create top level Actor with ActorSystem
// ActorSystem is a heavy object: create only one per application
val system = ActorSystem("mySystem")
val myActor1 = system.actorOf(props, "myactor")
}
}
-
ActorSystem是重量级组件，一个应用最好只创建一个。他创建的Actor都是top level的 -
ActorContext用于在Actor内部创建子actor -
actorOf返回的是immutable的ActorRef，他是对actor的引用。 -
actorOf的第二个参数是对actor的标示，该标示不能为空，不能以$开头，但是可以包含URL encode的字符
Best Practice
- 在class的companion中创建
props方法，返回对应actor class的Props对象 - 在class的companion中创建
case class或case object，用于枚举actor的信息类型
- Dependency Injection可以用来解决创建actor的依赖问题，具体还没研究。
Actor API
常用的api有如下几个
-
self，返回对自己的引用 -
sender，返回给当前actor发信息的引用，可以通过sender() ! "hello"回复信息。如果当前actor是被其他actor调用，则sender为调用的actor，如果不是actor调用，则sender为默认的deadLetteractor。 -
supervisorStrategy，可被用户重写，以实现对子actor的监控策略 -
context，ActorContext
还有一些lifecycle hook如下
def preStart(): Unit = ()
def postStop(): Unit = ()
def preRestart(reason: Throwable, message: Option[Any]): Unit = {
context.children foreach { child ?
context.unwatch(child)
context.stop(child)
}
postStop()
}
def postRestart(reason: Throwable): Unit = {
preStart()
}
Actor Selection (local/remote)
当actor由ActorContext一级级往下创建时，就会形成一个类似目录的父子关系，如/parent/child/grandchild。其中可以是相对路径(../myActor)或者绝对路径(/user/myActor)。
- 绝对路径的起点是
/user
import akka.actor.{Actor, ActorSystem, Props}
import akka.util.Timeout
import scala.concurrent.duration._
class MyActor extends Actor {
override def receive: Receive = {
case MyActor.Greeting => {
sender() ! "hello"
}
}
}
object MyActor {
case object Greeting
def props(): Props = {
Props(new MyActor())
}
def main(args: Array[String]): Unit = {
import akka.pattern.ask
implicit val timeout = Timeout(5 seconds)
val system = ActorSystem("mySys")
system.actorOf(MyActor.props(), "myactor")
import system.dispatcher
system.actorSelection("/user/myactor").resolveOne().onSuccess {
case ma => {
val future = ask(ma, MyActor.Greeting).mapTo[String]
future onSuccess {
case result => println(result)
}
}
}
Thread.sleep(1000)
system.terminate()
}
}
// output
// hello
具体参考：http://doc.akka.io/docs/akka/2.4/scala/actors.html#Identifying_Actors_via_Actor_Selection
Become/Unbecome
become(PartialFunction[Any, Unit])会用参数中的行为替换当前的行为，而unbecome()会还原成上一个行为。
这一对方法相当于压栈/出栈，所以需要注意栈溢出。
参考：http://doc.akka.io/docs/akka/2.4/scala/actors.html#Become_Unbecome
Send Message
可以通过下面两个方法发送信息：
-
!，fire-forget，即tell。这种方式不会被block住 -
?，send-and-receive-future，即ask。他会返回一个类似javaFuture的引用。信息的接受者必须通过!返回消息以结束Future。ask操作也需要创建一个handler来接受回复，以及一个timeout来处理超时后的资源回收。
timeout可以分两种方式定义：
显式：
import scala.concurrent.duration._
import akka.pattern.ask
val future = myActor.ask("hello")(5 seconds)
隐式：
import scala.concurrent.duration._
import akka.util.Timeout
import akka.pattern.ask
implicit val timeout = Timeout(5 seconds)
val future = myActor ? "hello"
同步ask
import akka.actor.{Actor, ActorSystem, Props}
import akka.pattern.ask
import akka.util.Timeout
import scala.concurrent.duration._
import scala.concurrent.Await
class MyActor extends Actor {
override def receive: Receive = {
case MyActor.Greeting => {
sender() ! "hello"
}
}
}
object MyActor {
case object Greeting
def props(): Props = {
Props(new MyActor())
}
def main(args: Array[String]): Unit = {
val mySystem = ActorSystem("mySys")
val ma = mySystem.actorOf(MyActor.props(), "myactor")
implicit val timeout = Timeout(5 seconds)
// use asInstanceOf
val future = ma ? MyActor.Greeting
val reply = Await.result(future, timeout.duration).asInstanceOf[String] // this will block until future finishes
println("waiting")
println(reply)
println("I m here")
// using mapTo is better
// val future = ask(ma, MyActor.Greeting).mapTo[String]
// val reply = Await.result(future, timeout.duration)
mySystem.terminate()
}
}
// output:
// waiting
// hello
// I m here
- Await.result将会block住，知道future结束，或者timeout
- Actor返回的future是Future[Any]，所以这里需要
asInstanceOf。或者通过mapTo返回一个指定类型的Future
Forward Message
可以通过target forward message来转发消息，这样的话，中间actor就相当与一个router
Receive Message
通过实现akka.actor.Actor的receive方法来接受信息。该方法返回PartialFunction，如case/match
type Receive = PartialFunction[Any, Unit]
def receive: Actor.Receive
Stop Actor
通过ActorSystem和ActorContext的stop方法可以停止一个actor，stop方法是异步的。
一个actor执行stop分三个步骤：
- actor挂起自己的mailbox，不再处理消息，并且给自己的子actor发送stop命令；
- actor会开始等待子actor的停止通知，直到所有的actor都停止；
- 最终停止自己
- 如果某个子actor没能停止成功，则停止过程会被stuck。
ActorSystem.terminate会停止ActorSystem
graceful stop：http://doc.akka.io/docs/akka/2.4/scala/actors.html#Graceful_Stop
Future
future是用来获取同步操作结果的，类似java
mapTo
如上面的例子，可以通过mapTo将Future[Any]转换成Future[String]等其他类型。
pipeTo
可以将Future的结果转发给一个Actor
import akka.pattern.pipe
import mySystem.dispatcher // ExecutionContext
future pipeTo ma
// or
pipe(future) to ma
-
ExecutionContext是Future运行需要的环境，类似java的Executor，每个Actor都被配置了MessageDispatcher，他也是一个ExecutionContext。所以可以通过import他来提供：
import mySystem.dispatcher // the created ActorSystem
// or
import myContext.dispatcher // the Actor's ActorContext
也可以自己创建：
import scala.concurrent.{ ExecutionContext, Promise }
implicit val ec = ExecutionContext.fromExecutorService(yourExecutorServiceGoesHere)
// Do stuff with your brand new shiny ExecutionContext
val f = Promise.successful("foo")
// Then shut your ExecutionContext down at some
// appropriate place in your program/application
ec.shutdown()
Use Directly
直接创建使用
import scala.concurrent.Await
import scala.concurrent.Future
import scala.concurrent.duration._
val future = Future {
"Hello" + "World"
}
future foreach println
Callback
future支持设置对结果的handler
object Main {
def main(args: Array[String]): Unit = {
import scala.concurrent.Future
import scala.util.Success
import scala.util.Failure
import scala.concurrent.ExecutionContext.Implicits.global
val future = Future {
val result = "hello"
result + "world"
}
future onSuccess {
case "hello" => println("sucess")
case x => println("success: " + x)
}
future onSuccess {
case "hello" => println("sucess")
case x => println("success: " + x)
}
future onFailure {
case ise: IllegalStateException if ise.getMessage == "OHNOES" =>
case e: Exception =>
}
future onComplete {
case Success(result) => println("complete with success: " + result)
case Failure(exception) => println("complete with failure: " + exception)
}
Thread.sleep(1000)
}
}
// output:
// complete with success: helloworld
// success: helloworld
// success: helloworld
Future in Order
通过andThen实现future序列，前面的future的结果会陆续作为后续future的输入
object Main {
def main(args: Array[String]): Unit = {
import scala.concurrent.Future
import scala.util.Success
import scala.util.Failure
import scala.concurrent.ExecutionContext.Implicits.global
Future {
val result = "hello"
result + "world"
} andThen {
case Success(result) => println("complete with success: " + result)
case Failure(exception) => println("complete with failure: " + exception)
} andThen {
case result => println("here: " + result)
}
Thread.sleep(1000)
}
}
// output:
// complete with success: helloworld
// here: Success(helloworld)
flow
fallbackTo返回一个新的future，当第一个future失败，则拿第二个future的结果
object Main {
def main(args: Array[String]): Unit = {
import scala.concurrent.Future
import scala.util.Success
import scala.concurrent.ExecutionContext.Implicits.global
case class CustomException(message: String = "", cause: Throwable = null)
extends Exception(message, cause)
val future1 = Future {
throw new CustomException("whatever")
}
val future2 = Future {
"hello"
}
val future = future1 fallbackTo future2
future onComplete {
case Success(result) => println(result)
}
Thread.sleep(1000)
}
}
// outut:
// hello
zip返回一个新的future，可以将两个future的结果绑定到一个tuple
object Main {
def main(args: Array[String]): Unit = {
import scala.concurrent.Future
import scala.concurrent.ExecutionContext.Implicits.global
val future1 = Future {
"hello"
}
val future2 = Future {
"world"
}
val future = future1 zip future2
future foreach println
Thread.sleep(1000)
}
}
// output
// (hello,world)
Exception
recover可以catch future抛出的异常
object Main {
def main(args: Array[String]): Unit = {
import scala.concurrent.Future
import scala.util.Success
import scala.concurrent.ExecutionContext.Implicits.global
case class CustomException(message: String = "", cause: Throwable = null)
extends Exception(message, cause)
val futureFail = Future {
throw new CustomException("whatever")
} recover {
case e => "catch: " + e
}
futureFail onComplete {
case Success(result) => println(result)
}
Thread.sleep(1000)
val futureSucc = Future {
"ok"
} recover {
case e => "catch: " + e
}
futureSucc onComplete {
case Success(result) => println(result)
}
Thread.sleep(1000)
}
}
// output
// catch: com.study.concurrency.actor.future.Main$CustomException$3: whatever
// ok
也可以用recoverWith返回一个新的future
object Main {
def main(args: Array[String]): Unit = {
import scala.concurrent.Future
import scala.util.Success
import scala.concurrent.ExecutionContext.Implicits.global
case class CustomException(message: String = "", cause: Throwable = null)
extends Exception(message, cause)
val futureFail = Future {
throw new CustomException("whatever")
} recoverWith {
case e => Future.failed[Int](new CustomException("fail")) // use method of Future companion to create a future
case _ => Future.successful("succ")
}
futureFail onComplete {
case Success(result) => println(result)
}
Thread.sleep(1000)
}
}
After
after可以给future一个timeout，超时返回指定的值
// TODO after is unfortunately shadowed by ScalaTest, fix as part of #3759
// import akka.pattern.after
val delayed = akka.pattern.after(200 millis, using = system.scheduler)(Future.failed(
new IllegalStateException("OHNOES")))
val future = Future { Thread.sleep(1000); "foo" }
val result = Future firstCompletedOf Seq(future, delayed)
