scala akka actor一种并发模型（四）

import akka.dispatch.RequiresMessageQueue
import akka.dispatch.BoundedMessageQueueSemantics

class MyBoundedActor extends MyActor
  with RequiresMessageQueue[BoundedMessageQueueSemantics]

bounded-mailbox {
  mailbox-type = "akka.dispatch.BoundedMailbox"
  mailbox-capacity = 1000
  mailbox-push-timeout-time = 10s
}

akka.actor.mailbox.requirements {
  "akka.dispatch.BoundedMessageQueueSemantics" = bounded-mailbox
	}

my-dispatcher {
  mailbox-requirement = org.example.MyInterface
}

akka.actor.default-mailbox {
  mailbox-type = "akka.dispatch.SingleConsumerOnlyUnboundedMailbox"
}

// 在这种情况下，我们从UnboundedStablePriorityMailbox继承
// and seed it with the priority generator
class MyPrioMailbox(settings: ActorSystem.Settings, config: Config)
  extends UnboundedStablePriorityMailbox(
	  // 创建一个新的PriorityGenerator，更低的prio意味着更重要
	  PriorityGenerator {
		  // 'highpriority messages should be treated first if possible
		  case 'highpriority => 0

		  // 'lowpriority messages should be treated last if possible
		  case 'lowpriority => 2

		  // PoisonPill when no other left
		  case PoisonPill => 3

		  // We default to 1, which is in between high and low
		  case otherwise => 1
	  })


// We create a new Actor that just prints out what it processes
class Logger extends Actor {
	self ! 'lowpriority
	self ! 'lowpriority
	self ! 'highpriority
	self ! 'pigdog
	self ! 'pigdog2
	self ! 'pigdog3
	self ! 'highpriority
	self ! PoisonPill

	def receive = {
		case x => println(x.toString)
	}
}

	val a = system.actorOf(Props(classOf[Logger]).withDispatcher(
		"prio-dispatcher"))
	/*输出
'highpriority
'highpriority
'pigdog
'pigdog2
'pigdog3
'lowpriority
'lowpriority
	*/

prio-dispatcher {
  type = Dispatcher
  executor = "fork-join-executor"
  fork-join-executor {
    parallelism-min = 2
    parallelism-factor = 2.0
    parallelism-max = 10
  }
  throughput = 100
  # 配置邮箱
  mailbox-type = "com.lightbend.akka.sample.MyPrioMailbox"
}

prio-mailbox {
  mailbox-type = "docs.dispatcher.DispatcherDocSpec$MyPrioMailbox"
  //Other mailbox configuration goes here
}

akka.actor.deployment {
  /priomailboxactor {
    mailbox = prio-mailbox
  }
}

import akka.actor.Props
val myActor = context.actorOf(Props[MyActor], "priomailboxactor")

control-aware-dispatcher {
  mailbox-type = "akka.dispatch.UnboundedControlAwareMailbox"
  //Other dispatcher configuration goes here
}

import akka.dispatch.ControlMessage
case object MyControlMessage extends ControlMessage

// We create a new Actor that just prints out what it processes
class Logger2 extends Actor {

	self ! 'foo
	self ! 'bar
	self ! MyControlMessage
	self ! PoisonPill

	def receive = {
		case x => println(x.toString)
	}
}

val a1 = system.actorOf(Props(classOf[Logger2]).withDispatcher(
		"control-aware-dispatcher"))
/*输出
MyControlMessage
'foo
'bar
*/

// 一个标记特质，被用来做mailbox requirements映射
trait MyUnboundedMessageQueueSemantics

import akka.actor.ActorRef
import akka.actor.ActorSystem
import akka.dispatch.Envelope
import akka.dispatch.MailboxType
import akka.dispatch.MessageQueue
import akka.dispatch.ProducesMessageQueue
import com.typesafe.config.Config
import java.util.concurrent.ConcurrentLinkedQueue

object MyUnboundedMailbox {
	// 消息队列的实现，需要混入 标记特质
	class MyMessageQueue extends MessageQueue
	  with MyUnboundedMessageQueueSemantics {

		private final val queue = new ConcurrentLinkedQueue[Envelope]()

		override def enqueue(receiver: ActorRef, handle: Envelope): Unit =
			queue.offer(handle)
		override def dequeue(): Envelope = queue.poll()
		override def numberOfMessages: Int = queue.size
		override def hasMessages: Boolean = !queue.isEmpty
		override def cleanUp(owner: ActorRef, deadLetters: MessageQueue) {
			while (hasMessages) {
				deadLetters.enqueue(owner, dequeue())
			}
		}
	}
}

// 这是邮箱消息队列的是创建工厂
class MyUnboundedMailbox extends MailboxType
  with ProducesMessageQueue[MyUnboundedMailbox.MyMessageQueue] {

	import MyUnboundedMailbox._

	// 这个构造函数签名必须存在，它将被Akka调用
	def this(settings: ActorSystem.Settings, config: Config) = {
		// 在这添加自己的初始化代码
		this()
	}

	// 创建消息队列的方法
	final override def create(
	                           owner: Option[ActorRef],
	                           system: Option[ActorSystem]): MessageQueue ={
		println("创建了一个消息队列")
		new MyMessageQueue()
	}
}

## 自定义邮箱
custom-dispatcher {
  type = Dispatcher
  executor = "fork-join-executor"
  fork-join-executor {
    parallelism-min = 2
    parallelism-factor = 2.0
    parallelism-max = 10
  }
  throughput = 100
  # 配置邮箱
  mailbox-requirement =
    "com.lightbend.akka.sample.MyUnboundedMessageQueueSemantics"
}

# 配置邮箱类型
custom-dispatcher-mailbox {
  mailbox-type = "com.lightbend.akka.sample.MyUnboundedMailbox"
}

# MyUnboundedMessageQueueSemantics trait 的创建工厂为
# MyUnboundedMailbox ，他将 创建一个 消息队列 with MyUnboundedMessageQueueSemantics
akka.actor.mailbox.requirements {
  "com.lightbend.akka.sample.MyUnboundedMessageQueueSemantics" =
    custom-dispatcher-mailbox
}

//定义Actor时指定
class MySpecialActor extends MyActor
  with RequiresMessageQueue[MyUnboundedMessageQueueSemantics] {
}

//创建Actor时指定，执行上下文方式
system.actorOf(Props[MyActor].withDispatcher("custom-dispatcher"))
system.actorOf(Props[MySpecialActor])

package com.lightbend.akka.sample

import akka.actor.{Actor, ActorSystem, Props, Terminated}
import akka.routing.{ActorRefRoutee, RoundRobinRoutingLogic, Router}

object Worker {
	case object Work
}

class Worker extends Actor{
	import Worker._

	override def receive:Receive = {
		case Work => println("工作中")
	}
}

class Master extends Actor {
	var router:Router = {
		val routees = Vector.fill(5) {
			val r = context.actorOf(Props[Worker])
			context watch r
			ActorRefRoutee(r)
		}
		Router(RoundRobinRoutingLogic(), routees)
	}
	import Worker._

	override def receive:Receive = {
		case Work =>
			router.route(Work, sender())
		case Terminated(a) =>
			router = router.removeRoutee(a)
			val r = context.actorOf(Props[Worker])
			context watch r
			router = router.addRoutee(r)
	}
}

object RouterTest extends App {
	val system = ActorSystem("RouterTest")
	val master = system.actorOf(Props[Master], "master")
	master ! Worker.Work
	master ! Worker.Work
	master ! Worker.Work
}

akka.actor.deployment {
  /router1 {
    router = round-robin-pool
    nr-of-instances = 5
  }
}

val router1: ActorRef =
  context.actorOf(FromConfig.props(Props[Worker]), "router1")

val router2: ActorRef =
  context.actorOf(RoundRobinPool(5).props(Props[Worker]), "router2")

import akka.actor.{ Address, AddressFromURIString }
import akka.remote.routing.RemoteRouterConfig
val addresses = Seq(
  Address("akka.tcp", "remotesys", "otherhost", 1234),
  AddressFromURIString("akka.tcp://othersys@anotherhost:1234"))
val routerRemote = system.actorOf(
  RemoteRouterConfig(RoundRobinPool(5), addresses).props(Props[Echo]))

sender() ! x // replies will go to this actor

sender().tell("reply", context.parent) // replies will go back to parent
sender().!("reply")(context.parent) // alternative syntax (beware of the parens!)

val escalator = OneForOneStrategy() {
  case e ⇒ testActor ! e; SupervisorStrategy.Escalate
}
val router = system.actorOf(RoundRobinPool(1, supervisorStrategy = escalator).props(
  routeeProps = Props[TestActor]))

akka.actor.deployment {
  /router3 {
    router = round-robin-group
    routees.paths = ["/user/w1", "/user/w2", "/user/w3"]
  }
}

	system.actorOf(Props[Worker], "w1")
	system.actorOf(Props[Worker], "w2")
	system.actorOf(Props[Worker], "w3")

	val router3: ActorRef =
		system.actorOf(FromConfig.props(), "router3")

	router3 ! Worker.Work

val router4: ActorRef =
  context.actorOf(RoundRobinGroup(paths).props(), "router4")

akka.actor.deployment {
  /parent/router1 {
    router = round-robin-pool
    nr-of-instances = 5
  }
}

val router1: ActorRef =
  context.actorOf(FromConfig.props(Props[Worker]), "router1")

val router2: ActorRef =
  context.actorOf(RoundRobinPool(5).props(Props[Worker]), "router2")

akka.actor.deployment {
  /parent/router3 {
    router = round-robin-group
    routees.paths = ["/user/workers/w1", "/user/workers/w2", "/user/workers/w3"]
  }
}

val router3: ActorRef =
  context.actorOf(FromConfig.props(), "router3")

val paths = List("/user/workers/w1", "/user/workers/w2", "/user/workers/w3")
val router4: ActorRef =
  context.actorOf(RoundRobinGroup(paths).props(), "router4")

akka.actor.deployment {
  /parent/router5 {
    router = random-pool
    nr-of-instances = 5
  }
}

akka.actor.deployment {
  /parent/router7 {
    router = random-group
    routees.paths = ["/user/workers/w1", "/user/workers/w2", "/user/workers/w3"]
  }
}

val router5: ActorRef =
  context.actorOf(FromConfig.props(Props[Worker]), "router5")

val router6: ActorRef =
  context.actorOf(RandomPool(5).props(Props[Worker]), "router6")

val router7: ActorRef =
  context.actorOf(FromConfig.props(), "router7")

val paths = List("/user/workers/w1", "/user/workers/w2", "/user/workers/w3")
val router8: ActorRef =
  context.actorOf(RandomGroup(paths).props(), "router8")

import akka.routing.Broadcast
router4 ! Broadcast(Worker.Work)

import akka.actor.PoisonPill
router ! PoisonPill

import akka.actor.PoisonPill
import akka.routing.Broadcast
router ! Broadcast(PoisonPill)

akka.actor.deployment {
  /parent/router29 {
    router = round-robin-pool
    resizer {
      lower-bound = 2
      upper-bound = 15
      messages-per-resize = 100
    }
  }
}

val router29: ActorRef =
  context.actorOf(FromConfig.props(Props[Worker]), "router29")

val resizer = DefaultResizer(lowerBound = 2, upperBound = 15)
val router30: ActorRef =
  context.actorOf(
    RoundRobinPool(5, Some(resizer)).props(Props[Worker]),
    "router30")

akka.actor.deployment {
  /parent/router31 {
    router = round-robin-pool
    optimal-size-exploring-resizer {
      enabled = on
      action-interval = 5s
      downsize-after-underutilized-for = 72h
    }
  }
}

akka.actor.deployment {
  /poolWithDispatcher {
    router = random-pool
    nr-of-instances = 5
    pool-dispatcher {
      fork-join-executor.parallelism-min = 5
      fork-join-executor.parallelism-max = 5
    }
  }
}

val router: ActorRef = system.actorOf(
  // “head” router actor will run on "router-dispatcher" dispatcher
  // Worker routees will run on "pool-dispatcher" dispatcher
  RandomPool(5, routerDispatcher = "router-dispatcher").props(Props[Worker]),
  name = "poolWithDispatcher")

State(S) x Event(E) -> Actions (A), State(S’)

import akka.actor.{ ActorRef, FSM }
import scala.concurrent.duration._

// Buncher 接收可以接收的消息 events
final case class SetTarget(ref: ActorRef)
final case class Queue(obj: Any)
case object Flush

// Buncher 发送的消息 events
final case class Batch(obj: collection.immutable.Seq[Any])

// Buncher 内部状态
sealed trait State
case object Idle extends State //空闲
case object Active extends State //激活

// Buncher 内部数据
case object Uninitialized extends Data //未初始化状态的数据
sealed trait Data
final case class Todo(target: ActorRef, queue: collection.immutable.Seq[Any]) extends Data //能接受的数据

class Buncher extends FSM[State, Data] {

	startWith(Idle, Uninitialized) //初始化状态

	//在空闲状态的处理
	when(Idle) {
		case Event(SetTarget(ref), Uninitialized) =>
			stay using Todo(ref, Vector.empty) //创建一个数据
	}

	//当状态变换时
	onTransition {
		case Active -> Idle =>
			stateData match {
				case Todo(ref, queue) => ref ! Batch(queue)
				case _ => // nothing to do
			}
	}

	//激活状态的处理，超时时间为1秒
	when(Active, stateTimeout = 1.second) {
		case Event(Flush | StateTimeout, t: Todo) => //超时或者传递Flush命令时
			goto(Idle) using t.copy(queue = Vector.empty) //清空队列
	}

	//when无法处理的状态
	whenUnhandled {
		// common code for both states
		case Event(Queue(obj), t @ Todo(_, v)) =>
			goto(Active) using t.copy(queue = v :+ obj)

		case Event(e, s) =>
			log.warning("received unhandled request {} in state {}/{}", e, stateName, s)
			stay
	}

	initialize()
}

class ParentActor extends Actor {

	val buncherActor = context.actorOf(Props[Buncher], "buncher")
	var i = 0

	override def receive:Receive = {
		case "init" => buncherActor ! SetTarget(self)
		case "add" => buncherActor ! Queue(i); i+=1
		case "flush" => buncherActor ! Flush
		case Batch(queue) => println(queue)
	}
}


object FSMTest extends App{
	val system = ActorSystem("FSMTest")
	val p = system.actorOf(Props[ParentActor], "father")

	p ! "init"
	p ! "add"
	p ! "add"
	p ! "add"
	p ! "add"
	p ! "flush"
	p ! "add"
	p ! "add"
	p ! "add"
	p ! "flush"
}

when(<name>[, stateTimeout = <timeout>])(stateFunction)

startWith(state, data[, timeout])

whenUnhandled {
  case Event(x: X, data) =>
    log.info("Received unhandled event: " + x)
    stay
  case Event(msg, _) =>
    log.warning("Received unknown event: " + msg)
    goto(Error)
}

when(SomeState) {
  case Event(msg, _) =>
    goto(Processing) using (newData) forMax (5 seconds) replying (WillDo)
}

onTransition(handler)

setTimer(name, msg, interval, repeat)
cancelTimer(name)
isTimerActive(name)
	stop([reason[, data]])

"com.typesafe.akka" %% "akka-persistence" % "2.5.6"

"org.iq80.leveldb"            % "leveldb"          % "0.9"
"org.fusesource.leveldbjni"   % "leveldbjni-all"   % "1.8"

package com.lightbend.akka.sample.persistence

//#persistent-actor-example
import akka.actor._
import akka.persistence._
import com.typesafe.config.ConfigFactory

case class Cmd(data: String) //命令
case class Evt(data: String) //事件

//例子状态：持有事件列表
case class ExampleState(events: List[String] = Nil) {
	//更新事件列表
	def updated(evt: Evt): ExampleState = copy(evt.data :: events)
	//事件尺寸
	def size: Int = events.length
	override def toString: String = events.reverse.toString
}

class ExamplePersistentActor extends PersistentActor {
	//持久化id
	override def persistenceId = "sample-id-1"
	//内部维持的状态
	var state = ExampleState()
	//更新状态
	def updateState(event: Evt): Unit =
		state = state.updated(event)
	//状态内部事件列表的尺寸
	def numEvents =
		state.size

	//接收到回复状态的事件
	override val receiveRecover: Receive = {
		case evt: Evt => updateState(evt)
		case SnapshotOffer(_, snapshot: ExampleState) => state = snapshot
	}

	//接收到命令
	override val receiveCommand: Receive = {
		case Cmd(data) =>
			//持久化事件
			persist(Evt(s"${data}-${numEvents}"))(updateState)
			persist(Evt(s"${data}-${numEvents + 1}")) { event =>
				updateState(event)
				context.system.eventStream.publish(event)
			}
		case "snap" => saveSnapshot(state)
		case "print" => println(state)
	}

}

//#persistent-actor-example

object PersistentActorExample extends App {

	val config = ConfigFactory.load("persistence.conf")
	val system = ActorSystem("example",config)
	val persistentActor = system.actorOf(Props[ExamplePersistentActor], "persistentActor-4-scala")

	persistentActor ! Cmd("foo")
	persistentActor ! Cmd("baz")
	persistentActor ! Cmd("bar")
	persistentActor ! "snap"
	persistentActor ! Cmd("buzz")
	persistentActor ! "print"

	Thread.sleep(10000)
	system.terminate()
}

akka.persistence.journal.plugin = "akka.persistence.journal.leveldb"
akka.persistence.snapshot-store.plugin = "akka.persistence.snapshot-store.local"

akka.persistence.journal.leveldb.dir = "target/example/journal"
akka.persistence.snapshot-store.local.dir = "target/example/snapshots"

# DO NOT USE THIS IN PRODUCTION !!!
# See also https://github.com/typesafehub/activator/issues/287
akka.persistence.journal.leveldb.native = false

  "com.typesafe.akka"          %% "akka-persistence" % "2.5.4",
  "org.iq80.leveldb"            % "leveldb"          % "0.7",
  "org.fusesource.leveldbjni"   % "leveldbjni-all"   % "1.8"

List(foo-0, foo-1, baz-2, baz-3, bar-4, bar-5, buzz-6, buzz-7)

List(foo-0, foo-1, baz-2, baz-3, bar-4, bar-5, buzz-6, buzz-7, foo-8, foo-9, baz-10, baz-11, bar-12, bar-13, buzz-14, buzz-15)

akka.persistence.max-concurrent-recoveries = 50

override def recovery =
  Recovery(fromSnapshot = SnapshotSelectionCriteria.None)

override def recovery = Recovery(toSequenceNr = 457L)

override def recovery = Recovery.none

package com.lightbend.akka.sample

import akka.actor.ActorSystem
import akka.testkit.{ImplicitSender, TestActors, TestKit}
import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}

class TestKitTest()
	extends TestKit(ActorSystem("MySpec"))
	with ImplicitSender
	with WordSpecLike with Matchers with BeforeAndAfterAll {

	override def afterAll {
		TestKit.shutdownActorSystem(system)
	}

	"An Echo actor" must {

		"send back messages unchanged" in {
			val echo = system.actorOf(TestActors.echoActorProps)
			echo ! "hello world"
			expectMsg("hello world")
		}

	}
}

val hello: String = expectMsg("hello")
val any: String = expectMsgAnyOf("hello", "world")
val all: immutable.Seq[String] = expectMsgAllOf("hello", "world")
val i: Int = expectMsgType[Int]
expectNoMsg(200.millis)
val two: immutable.Seq[AnyRef] = receiveN(2)

package com.lightbend.akka.sample

import akka.actor.ActorSystem
import akka.testkit.{ImplicitSender, TestActors, TestKit}
import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}

class TestKitTest()
	extends TestKit(ActorSystem("MySpec"))
	with ImplicitSender //混入隐式发送者，直接通过!发送消息，将隐式设置发送者
	with WordSpecLike with Matchers with BeforeAndAfterAll {

	override def afterAll {
		TestKit.shutdownActorSystem(system)
	}

	"An Echo actor" must {

		"send back messages unchanged" in {
			val echo = system.actorOf(TestActors.echoActorProps)
			echo ! "hello world"
			val ret1 = expectMsg("hello world")
			println(ret1)

			echo ! "hello"

			val ret2 = expectMsgPF(){
				case "hello" => "hello"
			}
			println(ret2)



			echo ! "123"
			val ret3 = expectMsgClass(classOf[String])
			println(ret3)

			echo ! Vector(1, 2, 3)
			val ret4 = expectMsgType[Vector[Int]]
			println(ret4)

			echo ! "345"
			val ret5 = expectMsgAnyOf("123","345")
			println(ret5)

			echo ! 1
			val ret6 = expectMsgAnyClassOf(classOf[Int], classOf[String])
			println(ret6)

			echo ! "123"
			echo ! "345"
			echo ! "456"
			val ret7 = expectMsgAllOf("123","345")
			println(ret7)

			echo ! 3
			val ret8 = expectMsgAllClassOf(classOf[String], classOf[Int])
			println(ret8)

			echo ! 1
			echo ! "Stt"
			val ret9 = expectMsgAllConformingOf(classOf[String], classOf[Int])
			println(ret9)

			echo ! 10
			echo ! 11
			val ret10 = receiveN(2)
			println(ret10)
		}

	}
}