Akka 是不是原生支持集成模式？

Posted 2023-02-25

【中文标题】Akka 是不是原生支持集成模式？

【英文标题】：Does Akka natively support Integration Patterns?Akka 是否原生支持集成模式？

【发布时间】：2015-03-26 12:21:28

【问题描述】：

我是 Akka 的新手，我想弄清楚它是否具有对 Enterprise Integration Patterns (EIP) 的内置支持，或者我是否需要将这种类型的路由/集成委托给像 Camel 这样的框架。

在我的用例中，我有一个从源（文件）读取二进制样本的参与者；这个演员被称为Sampler。 Sampler 将 Sample 实例（消息）传递到名为 SampleProcessors 的参与者字段。每个示例处理器对给定的Sample 执行不同的操作。根据处理Sample 的处理器的结果，它可能需要路由到1+ 个其他SampleProcessor，或者可能所有处理都已结束。根据确切的SampleProcessor 和给定Sample 的确切性质，Sample 可能需要多播到其他收件人SampleProcessors 的列表。

这一切对我来说就像骆驼。

所以我问：

Akka 是否具有对路由、广播、多播和其他 EIP 的内置支持（如果有，它们是什么以及它们记录在哪里）？ 或者，我应该尝试将actor系统与Camel集成，在这种情况下，会是什么样子？我知道有一个 Camel-Akka 组件，但我相信这只是为了将 Camel 总线与 Actor 系统集成（而我想要一个服务总线在我的 Actor 系统中） 或者，我应该在这里自己做 EIP/actor 布线吗？

【问题讨论】：

Does Akka obsolesce Camel?的可能重复

【参考方案1】：

Akka 本身不支持 EIP，但有多种实现方式。

无论如何，如果你想要一些方便的 DSL，有一个比 EIP 更好的主意 - 正如使用 GoF-patterns 所做的那样，你可以用函数组合 + Functors (map) 替换（实现）大多数 EIP 模式和单子 (flatMap)。换句话说，您可以将输入流视为无限集合。所以，

处理器变成函数 管道变成函子，比如val output1 = input.map(processor1).map(processor2)

路由器和过滤器变成... monads（filter 基于flatMap）：

val fork1 = output1.filter(routingCondition1).map(...)

val fork2 = output1.filter(routingCondition2).map(...)

拆分为flatMap:input.flatMap(x => Stream(x.submsg1, x.submsg2))

聚合器变成变态，又名fold（累加器通常应该有一些存储支持）

这种基于流的工作流已经为 Akka 实现了，它被称为Akka Streams，它是Reactive Streams 的实现，另见this 和that文章。

另一种选择是按原样使用 Akka，actor 保证顺序处理，因此您可以通过创建 actor 链来实现管道：

class Processor1(next: ActorRef) extends Actor 
   def receive = 
      case x if filterCondition => 
      case x => next ! process(x)
   


val processor2 = system.actorOf(Props[Processor2])
val processor1 = system.actorOf(Props[Processor1], processor2)

如果您需要路由 - 它只是两个“下一个”

class Router(next1: ActorRef, next2: ActorRef) extends Actor 
   def receive = 
      case x if filterCondition => 
      case x if cond1 => next1 ! process(x)
      case x if cond2 => next2 ! process(x)
   

如果您需要保证路线之间没有比赛 - 请参阅this answer。当然，你放弃了整个 DSL 的想法，直接使用演员。

附：是的，您仍然可以将 Camel 用于端点 - Akka 有一些 support 用于此目的。你可以使用 Akka 作为服务激活器。