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

	//请求查询所有温度
	final case class RequestAllTemperatures(requestId: Long)
	//返回温度
	final case class RespondAllTemperatures(requestId: Long, temperatures: Map[String, TemperatureReading])

	//设备所有可能的情况
	sealed trait TemperatureReading
	//可读的
	final case class Temperature(value: Double) extends TemperatureReading
	//不可读
	case object TemperatureNotAvailable extends TemperatureReading
	//设备已关闭
	case object DeviceNotAvailable extends TemperatureReading
	//超时
	case object DeviceTimedOut extends TemperatureReading

package com.lightbend.akka.sample

import akka.actor.{Actor, ActorLogging, ActorRef, Cancellable, Props, Terminated}

import scala.concurrent.duration.FiniteDuration

//设备组查询actor伴生对象
object DeviceGroupQuery {
	//超时
	case object CollectionTimeout

	def props(
	           actorToDeviceId: Map[ActorRef, String],
	           requestId: Long,
	           requester: ActorRef,
	           timeout: FiniteDuration): Props = {
		Props(new DeviceGroupQuery(actorToDeviceId, requestId, requester, timeout))
	}
}

class DeviceGroupQuery(
                        actorToDeviceId: Map[ActorRef, String], //设备actor容器
                        requestId: Long, //请求号
                        requester: ActorRef, //请求者
                        timeout: FiniteDuration) extends Actor with ActorLogging {//超时时间
	import DeviceGroupQuery._
	import context.dispatcher

	//设置查询超时，查过这时间，receive将收到一个CollectionTimeout
	val queryTimeoutTimer: Cancellable = context.system.scheduler.scheduleOnce(timeout, self, CollectionTimeout)

	override def preStart(): Unit = {
		//观察所有设备actor的变化
		actorToDeviceId.keysIterator.foreach { deviceActor =>
			context.watch(deviceActor)
			deviceActor ! Device.ReadTemperature(0) //发送读取温度的消息
		}
	}

	override def postStop(): Unit = {
		queryTimeoutTimer.cancel()
	}

	override def receive: Receive =
		waitingForReplies(
			Map.empty,
			actorToDeviceId.keySet)

	//返回一个偏函数
	def waitingForReplies( //到目前为止已经回复的设备
	                       repliesSoFar: Map[String, DeviceGroup.TemperatureReading],
	                       stillWaiting: Set[ActorRef]): Receive = { //仍然在等待的设备
		case Device.RespondTemperature(0, valueOption) => //收到返回的温度
			val deviceActor = sender()
			val reading = valueOption match { //对温度进行封装
				//设备状态：温度有值
				case Some(value) => DeviceGroup.Temperature(value)
				//设备状态：没有值
				case None => DeviceGroup.TemperatureNotAvailable
			}
			//处理
			receivedResponse(deviceActor, reading, stillWaiting, repliesSoFar)

		case Terminated(deviceActor) => //设备已经关闭
			//设备状态：设备不可访问
			receivedResponse(deviceActor, DeviceGroup.DeviceNotAvailable, stillWaiting, repliesSoFar)

		//设备查询超时
		case CollectionTimeout =>
			val timedOutReplies = //将所有超时（仍在等待的）设备的，设备状态：设置为超时
				stillWaiting.map { deviceActor =>
					val deviceId = actorToDeviceId(deviceActor)
					deviceId -> DeviceGroup.DeviceTimedOut
				}
			//返回所有查询的消息
			requester ! DeviceGroup.RespondAllTemperatures(requestId, repliesSoFar ++ timedOutReplies)
			context.stop(self) //此时停止查询actor
	}

	def receivedResponse(
	                      deviceActor: ActorRef, //当前响应的设备
	                      reading: DeviceGroup.TemperatureReading, //当前设备温度状态
	                      stillWaiting: Set[ActorRef], //仍然在等待的设备组（包括当前响应的设备）
	                    //到目前为止已经相应完成的设备
	                      repliesSoFar: Map[String, DeviceGroup.TemperatureReading]): Unit = {
		context.unwatch(deviceActor) //取消观察已经响应的设备
		val deviceId = actorToDeviceId(deviceActor) //获取相应响应设备的actor引用
		val newStillWaiting = stillWaiting - deviceActor //将设备从等待区移除

		val newRepliesSoFar = repliesSoFar + (deviceId -> reading) //将设备加入完成响应列表
		if (newStillWaiting.isEmpty) { //如果完成了所有的响应
			//发送结果
			requester ! DeviceGroup.RespondAllTemperatures(requestId, newRepliesSoFar)
			//停止自己
			context.stop(self)
		} else { //等待其他响应
			//重新设置receive
			context.become(waitingForReplies(newRepliesSoFar, newStillWaiting))
		}
	}

}

package com.lightbend.akka.sample

import akka.actor.{ActorSystem, PoisonPill}
import akka.testkit.{TestKit, TestProbe}
import org.scalatest.{BeforeAndAfterAll, FlatSpecLike, Matchers}

import scala.concurrent.duration._

class DeviceGroupQuerySpec(_system: ActorSystem) extends TestKit(_system)
	with Matchers with FlatSpecLike	with BeforeAndAfterAll {

	def this() = this(ActorSystem("DeviceGroupQuerySpec"))

	override def afterAll: Unit = {
		shutdown(system)
	}

	"Query" should "return temperature value for working devices" in {
		val requester = TestProbe()

		val device1 = TestProbe()
		val device2 = TestProbe()

		//创建一个查询器
		val queryActor = system.actorOf(DeviceGroupQuery.props(
			actorToDeviceId = Map(device1.ref -> "device1", device2.ref -> "device2"),
			requestId = 1,
			requester = requester.ref,
			timeout = 3.seconds))

		//设备期望收到的消息是Device.ReadTemperature(requestId = 0)
		device1.expectMsg(Device.ReadTemperature(requestId = 0))
		device2.expectMsg(Device.ReadTemperature(requestId = 0))

		//给queryActor发送消息，消息为Device.RespondTemperature(requestId = 0, Some(1.0))，发送者是设备
		queryActor.tell(Device.RespondTemperature(requestId = 0, Some(1.0)), device1.ref)
		queryActor.tell(Device.RespondTemperature(requestId = 0, Some(2.0)), device2.ref)

		//期望query发送给请求者的消息为
		requester.expectMsg(DeviceGroup.RespondAllTemperatures(
			requestId = 1,
			temperatures = Map(
				"device1" -> DeviceGroup.Temperature(1.0),
				"device2" -> DeviceGroup.Temperature(2.0))))
	}

	"Query" should "return TemperatureNotAvailable for devices with no readings" in {
		val requester = TestProbe()

		val device1 = TestProbe()
		val device2 = TestProbe()

		val queryActor = system.actorOf(DeviceGroupQuery.props(
			actorToDeviceId = Map(device1.ref -> "device1", device2.ref -> "device2"),
			requestId = 1,
			requester = requester.ref,
			timeout = 3.seconds))

		device1.expectMsg(Device.ReadTemperature(requestId = 0))
		device2.expectMsg(Device.ReadTemperature(requestId = 0))

		queryActor.tell(Device.RespondTemperature(requestId = 0, None), device1.ref)
		queryActor.tell(Device.RespondTemperature(requestId = 0, Some(2.0)), device2.ref)

		requester.expectMsg(DeviceGroup.RespondAllTemperatures(
			requestId = 1,
			temperatures = Map(
				"device1" -> DeviceGroup.TemperatureNotAvailable,
				"device2" -> DeviceGroup.Temperature(2.0))))
	}

	"Query" should "return temperature reading even if device stops after answering" in {
		val requester = TestProbe()

		val device1 = TestProbe()
		val device2 = TestProbe()

		val queryActor = system.actorOf(DeviceGroupQuery.props(
			actorToDeviceId = Map(device1.ref -> "device1", device2.ref -> "device2"),
			requestId = 1,
			requester = requester.ref,
			timeout = 3.seconds))

		device1.expectMsg(Device.ReadTemperature(requestId = 0))
		device2.expectMsg(Device.ReadTemperature(requestId = 0))

		queryActor.tell(Device.RespondTemperature(requestId = 0, Some(1.0)), device1.ref)
		queryActor.tell(Device.RespondTemperature(requestId = 0, Some(2.0)), device2.ref)
		device2.ref ! PoisonPill

		requester.expectMsg(DeviceGroup.RespondAllTemperatures(
			requestId = 1,
			temperatures = Map(
				"device1" -> DeviceGroup.Temperature(1.0),
				"device2" -> DeviceGroup.Temperature(2.0))))
	}

	"Query" should "return DeviceTimedOut if device does not answer in time" in {
		val requester = TestProbe()

		val device1 = TestProbe()
		val device2 = TestProbe()

		val queryActor = system.actorOf(DeviceGroupQuery.props(
			actorToDeviceId = Map(device1.ref -> "device1", device2.ref -> "device2"),
			requestId = 1,
			requester = requester.ref,
			timeout = 1.second))

		device1.expectMsg(Device.ReadTemperature(requestId = 0))
		device2.expectMsg(Device.ReadTemperature(requestId = 0))

		queryActor.tell(Device.RespondTemperature(requestId = 0, Some(1.0)), device1.ref)

		requester.expectMsg(DeviceGroup.RespondAllTemperatures(
			requestId = 1,
			temperatures = Map(
				"device1" -> DeviceGroup.Temperature(1.0),
				"device2" -> DeviceGroup.DeviceTimedOut)))
	}
}

class DeviceGroup(groupId: String) extends Actor with ActorLogging {
  var deviceIdToActor = Map.empty[String, ActorRef]
  var actorToDeviceId = Map.empty[ActorRef, String]
  var nextCollectionId = 0L

  override def preStart(): Unit = log.info("DeviceGroup {} started", groupId)

  override def postStop(): Unit = log.info("DeviceGroup {} stopped", groupId)

  override def receive: Receive = {
    // ... other cases omitted

    case RequestAllTemperatures(requestId) =>
      context.actorOf(DeviceGroupQuery.props(
        actorToDeviceId = actorToDeviceId,
        requestId = requestId,
        requester = sender(),
        3.seconds))
  }

}

"be able to collect temperatures from all active devices" in {
  val probe = TestProbe()
  val groupActor = system.actorOf(DeviceGroup.props("group"))

  groupActor.tell(DeviceManager.RequestTrackDevice("group", "device1"), probe.ref)
  probe.expectMsg(DeviceManager.DeviceRegistered)
  val deviceActor1 = probe.lastSender

  groupActor.tell(DeviceManager.RequestTrackDevice("group", "device2"), probe.ref)
  probe.expectMsg(DeviceManager.DeviceRegistered)
  val deviceActor2 = probe.lastSender

  groupActor.tell(DeviceManager.RequestTrackDevice("group", "device3"), probe.ref)
  probe.expectMsg(DeviceManager.DeviceRegistered)
  val deviceActor3 = probe.lastSender

  // Check that the device actors are working
  deviceActor1.tell(Device.RecordTemperature(requestId = 0, 1.0), probe.ref)
  probe.expectMsg(Device.TemperatureRecorded(requestId = 0))
  deviceActor2.tell(Device.RecordTemperature(requestId = 1, 2.0), probe.ref)
  probe.expectMsg(Device.TemperatureRecorded(requestId = 1))
  // No temperature for device3

  groupActor.tell(DeviceGroup.RequestAllTemperatures(requestId = 0), probe.ref)
  probe.expectMsg(
    DeviceGroup.RespondAllTemperatures(
      requestId = 0,
      temperatures = Map(
        "device1" -> DeviceGroup.Temperature(1.0),
        "device2" -> DeviceGroup.Temperature(2.0),
        "device3" -> DeviceGroup.TemperatureNotAvailable)))
}

package com.lightbend.akka.sample

import akka.actor.ActorSystem
import akka.testkit.{TestKit, TestProbe}
import org.scalatest.{BeforeAndAfterAll, FlatSpecLike, Matchers}
import scala.concurrent.duration._

class DeviceSpec (_system: ActorSystem) extends TestKit(_system)
	with Matchers with FlatSpecLike with BeforeAndAfterAll {

	def this() = this(ActorSystem("系统名"))

	override def afterAll: Unit = {
		shutdown(system)
	}

	"什么" should "怎么样" in {
		//测试内容
	}
}

//使用
import akka.actor.{Actor, ActorLogging, ActorRef, Cancellable, Props, Terminated}

import scala.concurrent.duration.FiniteDuration


//测试
import akka.actor.{ActorSystem, PoisonPill}
import akka.testkit.{TestKit, TestProbe}
import org.scalatest.{BeforeAndAfterAll, FlatSpecLike, Matchers}

import scala.concurrent.duration._

class ErrorActor extends Actor {
	override def preStart(): Unit = println("preStart执行")
	override def postStop(): Unit = println("postStop执行")
	override def preRestart(reason: Throwable, message: Option[Any]): Unit = {
		super.preRestart(reason, message)
		println("preRestart执行")
	}
	override def postRestart(reason: Throwable): Unit = {
		super.postRestart(reason)
		println("postRestart执行")
	}
	override def receive:Receive = {
		case _ => throw new Exception
	}
}

	val system = ActorSystem("testSystem")
	val childProps = Props(classOf[ErrorActor])
	val supervisor = BackoffSupervisor.props(
		Backoff.onStop(
			childProps,
			childName = "errorActor",
			minBackoff = 3.seconds,
			maxBackoff = 30.seconds,
			randomFactor = 0.2 // 增加了20％的“噪音”，以稍微改变间隔
		)//.withManualReset // 孩子必须将BackoffSupervisor.Reset发送给其父母
		  .withDefaultStoppingStrategy //停在任何异常抛出

	)

	val supervisorActor = system.actorOf(supervisor, name = "errorActorSupervisor")

	println("第一次异常")
	supervisorActor ! "123"

val supervisor = BackoffSupervisor.props(
  Backoff.onFailure(
    childProps,
    childName = "myEcho",
    minBackoff = 3.seconds,
    maxBackoff = 30.seconds,
    randomFactor = 0.2 // adds 20% "noise" to vary the intervals slightly
  ).withAutoReset(10.seconds) // reset if the child does not throw any errors within 10 seconds
    .withSupervisorStrategy(
      OneForOneStrategy() {
        case _: MyException => SupervisorStrategy.Restart
        case _ => SupervisorStrategy.Escalate
      }))

"akka://my-sys/user/service-a/worker1"                   // purely local
"akka.tcp://[email protected]:5678/user/service-b" // remote

# In this file you can override any option defined in the reference files.
# Copy in parts of the reference files and modify as you please.

akka {

  # Loggers to register at boot time (akka.event.Logging$DefaultLogger logs
  # to STDOUT)
  loggers = ["akka.event.slf4j.Slf4jLogger"]

  # Log level used by the configured loggers (see "loggers") as soon
  # as they have been started; before that, see "stdout-loglevel"
  # Options: OFF, ERROR, WARNING, INFO, DEBUG
  loglevel = "DEBUG"

  # Log level for the very basic logger activated during ActorSystem startup.
  # This logger prints the log messages to stdout (System.out).
  # Options: OFF, ERROR, WARNING, INFO, DEBUG
  stdout-loglevel = "DEBUG"

  # Filter of log events that is used by the LoggingAdapter before
  # publishing log events to the eventStream.
  logging-filter = "akka.event.slf4j.Slf4jLoggingFilter"

  actor {
    provider = "cluster"

    default-dispatcher {
      # Throughput for default Dispatcher, set to 1 for as fair as possible
      throughput = 10
    }
  }

  remote {
    # The port clients should connect to. Default is 2552.
    netty.tcp.port = 4711
  }
}