AKKA中的Event Stream

在《企业应用集成模式》一书中，定义了许多与消息处理有关的模式，其中运用最为广泛的模式为Publisher-Subscriber模式，尤其是在异步处理场景下。

基于Publisher-Subscriber模式，还可以根据不同的场景衍生出特殊的模式，例如针对一个Publisher和多个Subscriber，演化为Broadcast模式和Message Router模式。前者会将消息同时发送给所有的Subscriber，实现分布式的并行处理。例如针对订单处理的场景，当顾客下订单后，既需要生成订单，又需要通知库存准备发货，还需要通知卖方和买方。这些任务虽然存在事务的一致性，但基于BASE原则，可以通过补偿机制实现事务的最终一致性。于是，设计时可以将这些任务交给不同的Subscriber，当接收到消息后，同时对订单进行处理。至于Message Router，则需要引入的Router对传入的消息作出智能判断，从而将消息传递给真正感兴趣的Subscriber。这就好像发布者同时发布了不同的刊物，订阅者只订阅自己喜欢的刊物。

而消息总线（message bus）则通过引入总线来彻底解除Publisher与Subscriber之间的耦合，类似设计模式中的Mediator模式。总线就是Mediator，用以协调Publisher与Subscriber之间的关系。或者，我们也可以认为是两个Publisher-Subscriber的组合。对于Publisher而言，总线就是Subscriber；对于Subscriber而言，总线则成了Publisher。

AKKA提供的事件总线（Event Bus）可以看做是一种运用于特殊场景的消息总线，此时事件即为消息。它可以看做是Message Router模式的实现，提供了向多个Actor发送消息的基础设施，内含的Classifier作为分类器，用于分发消息时选择Subscriber，扮演了Message Router的角色。

在AKKA中，Event Bus被定义为trait，定义了基本的订阅、取消订阅、发布等对应的方法，代码如下所示：

trait EventBus {
  type Event
  type Classifier
  type Subscriber

  def subscribe(subscriber: Subscriber, to: Classifier): Boolean

  def unsubscribe(subscriber: Subscriber, from: Classifier): Boolean

  def unsubscribe(subscriber: Subscriber): Unit

  def publish(event: Event): Unit
}

根据AKKA官方文档的描述，Event为所有发布到该总线上的事件类型，Classifier是选择订阅者的分类器，Subscriber就是注册到该总线上的订阅者。它们均被定义为抽象的type，使得EventBus拥有最大的开放性。我们视情况而定去具体实现特定类型。例如针对Actor的EventBus，订阅者被指定为ActorRef类型：

trait ActorEventBus extends EventBus {
  type Subscriber = ActorRef
  protected def compareSubscribers(a: ActorRef, b: ActorRef) = a compareTo b
}

AKKA运用Cake Pattern，以trait类型定义了一些特化的分类器。例如：

trait ActorClassifier { this: EventBus ⇒
  type Classifier = ActorRef
}

trait PredicateClassifier { this: EventBus ⇒
  type Classifier = Event ⇒ Boolean
}

trait LookupClassification { this: EventBus ⇒

  protected final val subscribers = new Index[Classifier, Subscriber](mapSize(), new Comparator[Subscriber] {
    def compare(a: Subscriber, b: Subscriber): Int = compareSubscribers(a, b)
  })

  protected def mapSize(): Int

  protected def compareSubscribers(a: Subscriber, b: Subscriber): Int

  protected def classify(event: Event): Classifier

  protected def publish(event: Event, subscriber: Subscriber): Unit

  def subscribe(subscriber: Subscriber, to: Classifier): Boolean = subscribers.put(to, subscriber)

  def unsubscribe(subscriber: Subscriber, from: Classifier): Boolean = subscribers.remove(from, subscriber)

  def unsubscribe(subscriber: Subscriber): Unit = subscribers.removeValue(subscriber)

  def publish(event: Event): Unit = {
    val i = subscribers.valueIterator(classify(event))
    while (i.hasNext) publish(event, i.next())
  }
}

除此之外，还有诸如SubchannelClassification、ScanningClassification、ActorClassification等分类器，都采用了同样的方式定义为trait。这样就便于继承EventBus的类进行trait的混入，例如EventStream的定义：

class EventStream(private val debug: Boolean = false) extends LoggingBus with SubchannelClassification {
  type Event = AnyRef
  type Classifier = Class[_]
}

由于LoggingBus继承自ActorEventBus，故而EventStream中的Subscriber类型被定义为ActorRef。然后在EventStream中又重写了Event和Classfier类型，分别为AnyRef和Class，这说明任何Java引用对象都可以作为事件，而分类的依据则为Event的类型。

type Subscriber = ActorRef
type Event = AnyRef
type Classifier = Class[_]

EventStream继承了SubchannelClassification。在其中维持了订阅者列表，虽然该订阅列表类型为SubclassifiedIndex，不过我们可以将其简单地视为一个Map（实际情况更复杂，因为它实际上维护了分类的层级）：

trait SubchannelClassification {
  this: EventBus ⇒

  private lazy val subscriptions = new SubclassifiedIndex[Classifier, Subscriber]()
}

由于此时的Classifier在EventStream中被定义为Class[_]，Subscriber为ActorRef，因此subscripions相当于被定义为：

private lazy val subscriptions = new SubclassifiedIndex[Class[_], ActorRef]

当Actor出现故障，从而使得消息被转发给dead letter时，我们可能需要侦听这些死信，并对它们进行处理。则AKKA的做法就是通过EventStream来进行订阅：

class DeadLetterListener extends Actor {
  def receive = {
    case DeadLetter(msg, from, to) =>
      println(s"${System.currentTimeMillis()}: from $from to $to with message $msg.")
  }
}

val listener = system.actorOf(Props[DeadLetterListener], "listener")
system.eventStream.subscribe(listener, classOf[DeadLetter])

结合SubchannelClassification，我们来分析其执行过程。首先，它通过subscribe方法将DeadLetterListener的actor引用对象以及事件类型DeadLetter注册到SubchannelClassification中的subscriptions。当ActorSystem的任意actor发出DeadLetter时，就会触发EventStream的publish()方法（实际上是调用了SubchannelClassification的publish()方法，进而调用EventStream的publish()方法）：

class EventStream(private val debug: Boolean = false) extends LoggingBus with SubchannelClassification {
  protected def publish(event: AnyRef, subscriber: ActorRef) = {
    if (subscriber.isTerminated) unsubscribe(subscriber)
    else subscriber ! event
  }
}

此时就会通过subscriber（此时为前面定义的DeadLetterListener)发送event（此时为DeadLetter对象），从而进入到DeadLetterListener的receive方法中，打印出我想要的消息。

通过EventStream还可以处理日志消息。AKKA自身也提供了默认的处理器，可以配置在application.conf文件中：

akka {
  event-handlers = ["akka.event.Logging$DefaultLogger"]
}

这个默认的日志处理器会订阅高于配置级别的日志事件类，例如将日志级别配置为Debug：

system.eventStream.setLogLevel(Logging.DebugLevel)

通过这样的配置，所有低于Debug级别的日志事件发生时，都不会被EventStream分发。