利用Actor实现管道过滤器模式

《基于Actor的响应式编程》计划分为三部分，第一部分剖析响应式编程的本质思想，为大家介绍何谓响应式编程（Reactive Programming）。第二部分则结合两个案例来讲解如何在AKKA中实现响应式编程。第三部分则是这个主题的扩展，在介绍Reactive Manifesto的同时，介绍进行响应式编程更为主流的ReactiveX框架。本文是第二部分的第一个案例。

《剖析响应式编程的本质》从Actor模型与响应式编程中找到彼此相配的特征；然而空口无凭，没有一点真凭实据，凭什么他们能立下海誓山盟、比翼双飞呢？

其实，Vaughn Vernon早就作了称职的月老，还为他们写了一本鸳梦奇缘，总结了如何利用Actor模型实现响应式编程的消息模式《Reactive Messaging Pattterns with the Actor Model》。如果阅读过《企业集成模式》(Enterprise Integration Patterns)一书，你会发现Vaughn的新书近乎于是《企业集成模式》中各种消息模式在AKKA中的Actor实现。

顺便吐槽一句，本书中文版的译名《响应式架构——消息模式Actor实现与Scala、AKKA应用集成》颇有标题党之嫌。整本书其实只是在相对低的层面讲解Actor对消息模式的实现，几乎没有牵涉到任何架构方面的知识。

例如，响应式编程通常会与CQRS以及Event Sourcing结合，但本书几乎没有涉猎。其实，Vaughn Vernon还是挺老实的，英文书名交代得也很清楚，翻译成中文，却莫名其妙地给书名添油加醋，有意误导消费者，实在不该。

当然，书还是好书，仍有阅读价值。

其实，我们说到Actor模型与响应式编程的相配，更大程度是因为Actor已经为响应式编程的编程要素提供了现成的基础设施。例如在AKKA之下进行响应式编程，我们几乎不用再考虑如何进行异步消息通信、状态切换、并发处理、并行处理，以及对Actor的监督和错误处理策略的实现。这在很大程度上使得我们可以从纷繁复杂的基础设施实现中解脱出来，而仅需要专注于考虑数据流转与业务流程之间的关系。

管道过滤器模式

谈到数据流（或者消息流），我们会想到一个经典的架构模式：管道过滤器模式。数据在管道中流动，每经过一个过滤器都会被对应的过滤器按照自己的处理逻辑进行处理，处理后的数据又被接着传递给下一个过滤器。

引入管道过滤器的一个好处是它可以使得每个过滤器之间都是解耦的，这使得我们可以很好地扩展过滤器，改变数据处理的流程，而不需要调整Provider端的代码。

在AKKA中，Actor之间可以通过ActorRef引用对象建立关联，这种抽象层面的弱依赖使得Actor彼此之间能够很好地解耦。不过，Actor之间还存在一条隐形依赖关系，它是由Actor所能处理的消息对象悄悄引入的。这些消息对象对于Actor，就好似Actor的接口，它表明了该Actor只能处理什么样的消息类型。一旦消息的结构发生改变，又或者希望Actor支持更多的消息，就需要修改Actor的定义与实现。

为了避免隐形依赖，我们可以将管道传递的数据定义为一个通用的消息类型，所有注册管道的过滤器处理的都是相同的流。在Provider端，我们实现的单个过滤器Actor，与其他过滤器之间是没有任何依赖关系的，我们也无需考虑数据处理的顺序，仅需要考虑自己的消息处理逻辑。

从这个角度看，一个Actor的设计与实现，应该尽可能遵循“单一职责原则”与“信息专家模式”。Udi Dahan在CQRS架构中曾经提出“自治组件”的概念，那么在Actor模型中，我们也应该尽可能做到让每个Actor对象自治。

在第一部分《剖析响应式编程的本质》中，我曾经提到：

我们几乎可以将所有业务处理流程都可以建模为数据流的形式。

下面我们就来看看一个订单处理流程的案例。这个案例来自前述Vaughn Vernon的著作《Reactive Messaging Pattterns with the Actor Model》：

一条订单消息进入系统，在为了完成购物操作处理完该条消息前，必须做一些预备工作。首先必须对这条订单消息进行解密，然后需要验证发送这条消息外部实体的资格，最后应确保这条订单消息不是之前收到消息的复制品。

我们可以将这些业务流程视为不同的职责，分解为：

• 对订单的部分数据进行解密（decryption）
  
  
• 对订单进行认证
  
• 对订单进行去重处理
  
• 处理订单
  

遵循单一职责原则，我们将这些职责分别交给对应的独立Actor来承担。例如认证订单：

class Authenticator(nextFilter: ActorRef) extends Actor with ActorLogging {
  def receive: Receive = {
    case message: ProcessIncomingOrder =>
      val text = new String(message.orderInfo)
      log.info(s"Authenticator: processing $text")
      val orderText = text.replace("(certificate)", "")
      nextFilter !  ProcessIncomingOrder(orderText.toCharArray.map(_.toByte))
  }
}

每个Actor会接收一个nextFilter的ActorRef对象，但它们是完全解耦的。Actor只专注于自己的职责，一旦处理完订单消息，就可以将处理后的消息传递给下一个Actor。这种“分而治之”的思想可以将复杂的事情变得更简单，开发者每次只需要考虑一个相对简单的职责，知识变少，利于理解。

过滤器之间的组合完全交给客户端，如下代码所示：

val orderManager = system.actorOf(Props[OrderManagementSystem], "OrderManagementSystem")
val deduplicator = system.actorOf(Props(new Deduplicator(orderManager)), "Deduplicator")
val authenticator = system.actorOf(Props(new Authenticator(deduplicator)), "Authenticator")
val decrypter = system.actorOf(Props(new Decrypter(authenticator)), "Decrypter")
val acceptance = system.actorOf(Props(new OrderAcceptanceEndpoint(decrypter)), "OrderAcceptanceEndpoint")
acceptance ! rawOrderBytes
acceptance ! rawOrderBytes

是否觉得似曾相似？倘若我们熟悉设计模式，会发现这一模式与“职责链模式”有着如孪生兄弟般的相似类结构。然而，二者的行为仍有些微差别，在经典的职责链模式中，一旦职责对象满足匹配条件时，会在履行该职责后中断处理并返回，而管道过滤器则会从起点一直“流动”到终点，若无意外，中途不会中断。

使用Actor实现管道过滤器模式，则又有所不同，业务的处理流程是在消息的跳转之间完成的，且每个消息的处理都是异步非阻塞的。