Scala學習筆記：Actor & Future

關于scala的Actor和經(jīng)常配合使用的Future

Actor

Creating Actor & best practice

  • 步驟一：實現(xiàn)traitakka.actor.Actor即可創(chuàng)建一個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的構造函數(shù)

  • 步驟三：通過akka.actor.ActorSystem或akka.actor.ActorContext創(chuàng)建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是重量級組件，一個應用最好只創(chuàng)建一個。他創(chuàng)建的Actor都是top level的
  • ActorContext用于在Actor內(nèi)部創(chuàng)建子actor
  • actorOf返回的是immutable的ActorRef，他是對actor的引用。
  • actorOf的第二個參數(shù)是對actor的標示，該標示不能為空，不能以$開頭，但是可以包含URL encode的字符

Best Practice

  1. 在class的companion中創(chuàng)建props方法，返回對應actor class的Props對象
  2. 在class的companion中創(chuàng)建case class或case object，用于枚舉actor的信息類型
  • Dependency Injection可以用來解決創(chuàng)建actor的依賴問題，具體還沒研究。

Actor API

常用的api有如下幾個

  1. self，返回對自己的引用
  2. sender，返回給當前actor發(fā)信息的引用，可以通過sender() ! "hello"回復信息。如果當前actor是被其他actor調(diào)用，則sender為調(diào)用的actor，如果不是actor調(diào)用，則sender為默認的deadLetter actor。
  3. supervisorStrategy，可被用戶重寫，以實現(xiàn)對子actor的監(jiān)控策略
  4. 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一級級往下創(chuàng)建時，就會形成一個類似目錄的父子關系，如/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])會用參數(shù)中的行為替換當前的行為，而unbecome()會還原成上一個行為。
這一對方法相當于壓棧/出棧，所以需要注意棧溢出。
參考：http://doc.akka.io/docs/akka/2.4/scala/actors.html#Become_Unbecome

Send Message

可以通過下面兩個方法發(fā)送信息：

  1. !，fire-forget，即tell。這種方式不會被block住
  2. ?，send-and-receive-future，即ask。他會返回一個類似java Future的引用。信息的接受者必須通過!返回消息以結束Future。ask操作也需要創(chuàng)建一個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來轉發(fā)消息，這樣的話，中間actor就相當與一個router

Receive Message

通過實現(xiàn)akka.actor.Actor的receive方法來接受信息。該方法返回PartialFunction，如case/match

type Receive = PartialFunction[Any, Unit]

def receive: Actor.Receive

Stop Actor

通過ActorSystem和ActorContext的stop方法可以停止一個actor，stop方法是異步的。
一個actor執(zhí)行stop分三個步驟：

  1. actor掛起自己的mailbox，不再處理消息，并且給自己的子actor發(fā)送stop命令；
  2. actor會開始等待子actor的停止通知，直到所有的actor都停止；
  3. 最終停止自己
  • 如果某個子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的結果轉發(fā)給一個Actor

import akka.pattern.pipe
import mySystem.dispatcher  // ExecutionContext

future pipeTo ma
// or
pipe(future) to ma
  • ExecutionContext是Future運行需要的環(huán)境，類似java的Executor，每個Actor都被配置了MessageDispatcher，他也是一個ExecutionContext。所以可以通過import他來提供：
import mySystem.dispatcher  // the created ActorSystem
// or
import myContext.dispatcher // the Actor's ActorContext

也可以自己創(chuàng)建：

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

直接創(chuàng)建使用

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實現(xiàn)future序列，前面的future的結果會陸續(xù)作為后續(xù)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，當?shù)谝粋€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)
最后編輯于 ：
?著作權歸作者所有,轉載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務。

相關閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容