import java.util.concurrent.{ExecutorService, Executors, TimeUnit}
import akka.actor.{Actor, ActorSystem, Props}
import akka.util.Timeout
import scala.concurrent.{Await, ExecutionContext}
import scala.concurrent.duration.Duration

 

Runnable

无返回值

  class TestActor extends Actor {
    def receive = {
      case arg => {
        println("got : " + arg)
        Thread.sleep(1000)
      }
    }
  }

同步调用

val system = ActorSystem("ActorSystem")
val actor = system.actorOf(Props(new TestActor), "TestActor")

actor ! "whatever"

Callable

有返回值

  class TestActor extends Actor {
    def receive = {
      case arg => {
        println("got : " + arg)
        Thread.sleep(1000)
        sender ! "hello : " + arg
      }
    }
  }

异步调用

    val system = ActorSystem("ActorSystem")
    val actor = system.actorOf(Props(new TestActor), "TestActor")

    implicit val timeout = Timeout(10000, TimeUnit.SECONDS)
    import akka.pattern._

    //1
    val feature = actor ? "whatever"
    while (!feature.isCompleted) Thread.sleep(1000)
    println(feature.isCompleted)
    if (feature.isCompleted) {println(feature.value.get.isSuccess + ", " + feature.value.get.get);}

    //2
    println(Await.result(feature, Duration.create(1, TimeUnit.SECONDS)))

更多

并发控制

  val ec = ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(100))
  val system = ActorSystem("ActorSystem", None, None, Option(ec))

定时

    val system = ActorSystem("ActorSystem")
    system.scheduler.schedule(Duration.create(1, TimeUnit.SECONDS), Duration.create(1, TimeUnit.SECONDS))({
      println("trigger : " + System.currentTimeMillis)
    })(ec)

 

注意Actor相当于java中的单实例单线程，可以通过多个Actor来控制并发