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

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

image.png

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

image.png

剖析响应式编程的本质》从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实现管道过滤器模式,则又有所不同,业务的处理流程是在消息的跳转之间完成的,且每个消息的处理都是异步非阻塞的。

相关文章
|
消息中间件 数据库 云计算
云计算设计模式(十五)——管道和过滤器模式
云计算设计模式(十五)——管道和过滤器模式 分解,执行复杂处理成一系列可重复使用分立元件的一个任务。这种模式可以允许执行的处理进行部署和独立缩放任务元素提高性能,可扩展性和可重用性。
1279 0
|
存储 SQL JSON
信创迁移适配实战-MySQL到达梦数据库DM8的数据迁移
信创迁移适配实战-MySQL到达梦数据库DM8的数据迁移
7868 0
信创迁移适配实战-MySQL到达梦数据库DM8的数据迁移
|
5月前
|
XML 算法 安全
详解RAG五种分块策略,技术原理、优劣对比与场景选型之道
RAG通过检索与生成结合,提升大模型在企业场景的准确性与安全性。分块策略是其核心,直接影响检索效果与生成质量。本文系统解析五种主流分块方法:固定大小、语义、递归、基于结构和基于LLM的分块,对比其优缺点及适用场景,助力构建高效、可信的RAG系统,尤其适用于金融、医疗等高精度领域。(239字)
|
9月前
|
存储 数据库
RAG分块技术全景图:5大策略解剖与千万级生产环境验证
本文深入解析RAG系统中的五大文本分块策略,包括固定尺寸、语义、递归、结构和LLM分块,探讨其工程实现与优化方案,帮助提升知识检索精度与LLM生成效果。
1318 1
|
机器学习/深度学习 移动开发 测试技术
YOLOv11改进策略【模型轻量化】| 替换骨干网络为MoblieNetV2,含模型详解和完整配置步骤
YOLOv11改进策略【模型轻量化】| 替换骨干网络为MoblieNetV2,含模型详解和完整配置步骤
786 13
YOLOv11改进策略【模型轻量化】| 替换骨干网络为MoblieNetV2,含模型详解和完整配置步骤
|
缓存 小程序 API
微信小程序网络请求与API调用:实现数据交互
本文深入探讨了微信小程序的网络请求与API调用,涵盖`wx.request`的基本用法、常见场景(如获取数据、提交表单、上传和下载文件)及注意事项(如域名配置、HTTPS协议、超时设置和并发限制)。通过一个简单案例,演示了如何实现小程序与服务器的数据交互。掌握这些技能将帮助你构建功能更丰富的应用。
|
存储 监控 安全
Elasticsearch 8.X 集群 SSL 证书到期了,怎么更换?
Elasticsearch 8.X 集群 SSL 证书到期了,怎么更换?
|
存储 缓存 算法
大文件 MD5 SHA 校验时间优化之路
【8月更文挑战第12天】处理大文件的MD5与SHA校验时,可通过选择高效算法实现、分块读取处理文件、利用多线程并行处理、采用硬件加速及缓存校验结果等方式优化校验时间。例如,使用性能良好的加密库如`pycryptodome`替代Python的标准`hashlib`库;分块读取文件并逐块计算哈希值,减少内存占用;利用多线程处理不同文件块;若条件允许,使用硬件加速如Intel AES-NI指令集;以及缓存重复校验的文件哈希值避免重算。这些策略可显著提高校验速度和系统效率。
1877 1

热门文章

最新文章