AKKA中的事件流

在《企业应用集成模式》一书中,定义了许多与消息处理有关的模式,其中运用最为广泛的模式为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
}

除此之外,还有诸如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()方法:

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分发。

原文发布于微信公众号 - 逸言(YiYan_OneWord)

原文发表时间:2015-09-25

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏挖掘大数据

大数据面试秘诀:30道hadoop面试真题和解析

近年来,大数据概念被炒的非常热,大数据公司也在快速的崛起,而人才的需求也越来越多。对于正在找大数据相关工作的同学们来说,面试时遇到什么问题才是他们最关心的。在下...

23410
来自专栏杨建荣的学习笔记

海量数据迁移之分区并行抽取(r2笔记53天)

在之前的章节中分享过一些数据迁移中并行抽取的细节,比如一个表T 很大,有500G的数据,如果开启并行抽取,默认数据库中并行的最大值为64,那么生成的dump文件...

2708
来自专栏小勇DW3

HashMap的resezi方法中尾部遍历出现死循环问题 Tail Traversing (多线程)

在看HashMap源码是看到了resize()的源代码,当时发现在将old链表中引用数据复制到新的链表中时,发现复制过程中时,源码是进行了反序,此时是允许反序存...

754
来自专栏AhDung

【C#】给无窗口的进程发送消息

一个winform程序,我希望它不能多开(但是如何防多开不是本文要讲的),那么在用户启动第二个实例的时候,作为第二个实例来说,大概可以有这么几种做法:

853
来自专栏偏前端工程师的驿站

Design Pattern: Observer Pattern

1. Brief                               一直对Observer Pattern和Pub/Sub Pattern有所混淆,下...

1867
来自专栏牛客网

百度云部门 C++面试

14)读套接口时候返回0,时候时候产生EAGIN。【EAGIN也不太清楚,知道又这个玩意,不知道具体的,应该直接说不知道】

972
来自专栏三丰SanFeng

Linux Kernel CMPXCHG函数分析

最近看到Linux Kernel cmpxchg的代码,对实现很不理解。上网查了内嵌汇编以及Intel开发文档,才慢慢理解了,记录下来以享和我一样困惑的开发者。...

29710
来自专栏乐沙弥的世界

Percona XtraDB Cluster Perfomance Schema Instrumentation

为了改进监控,Percona XtraDB集群实施了一个基础架构,将Galera仪器(mutexes, cond-variables, files, threa...

620
来自专栏Android 研究

OKHttp源码解析(四)--中阶之拦截器及调用链

那我们书接上文。上篇文章已经说明了OKHttp有两种调用方式,一种是阻塞的同步请求,一种是异步的非阻塞的请求。但是无论同步还是异步都会调用下RealCall的 ...

904
来自专栏影子

开始食用grpc(之二)

转载请注明出处:https://www.cnblogs.com/funnyzpc/p/9570992.html

1813

扫码关注云+社区