Поддерживает ли Akka шаблоны интеграции?

Я новичок в Akka и пытаюсь выяснить, есть ли в нем встроенная поддержка шаблонов корпоративной интеграции (EIP) или мне нужно делегировать этот тип маршрутизации/интеграции такой структуре, как Camel.

В моем случае у меня есть актор, который считывает двоичные образцы из источника (файла); этого актера зовут Sampler. Затем Sampler передает Sample экземпляров (сообщений) в поле акторов с именем SampleProcessors. Каждый образец процессора делает что-то свое для данного Sample. В зависимости от результата процессора, обрабатывающего Sample, его может потребоваться перенаправить на 1+ другой SampleProcessor, или, возможно, вся обработка завершена. В зависимости от точного SampleProcessor и точного характера данного Sample может потребоваться многоадресная рассылка Sample в список других получателей SampleProcessors.

Мне все это напоминает Camel.

Поэтому я спрашиваю:

  • Есть ли в Akka встроенная поддержка маршрутизации, широковещания, многоадресной рассылки и других EIP (если да, то что это такое и где они задокументированы)?
  • Или мне следует попытаться интегрировать систему акторов с Camel, и в таком случае, как это будет выглядеть? Я знаю, что есть компонент Camel-Akka, но я считаю, что он предназначен только для интеграции шины Camel с системой акторов (тогда как мне нужна служебная шина внутри моей системы акторов)
  • Или я должен просто сделать здесь свою собственную проводку EIP / актера?

actor apache-camel akka thread-synchronization enterprise-integration
person smeeb    schedule 27.01.2015    source источник
comment
возможный дубликат Устаревает ли Akka Camel?   -  person Petter Nordlander    schedule 28.01.2015

Ответы (1)


arrow_upward
4
arrow_downward

Akka изначально не поддерживает EIP, но есть несколько способов реализовать его.

В любом случае, если вам нужен удобный DSL, есть идея получше, чем EIP — как это делается с GoF-паттерны, вы можете заменить (реализовать) большинство EIP-паттернов функциональной композицией + Функторы (map) и Монады (flatMap). Другими словами, вы можете рассматривать входной поток как бесконечную коллекцию. Так,

  • процессоры становятся функциями
  • трубы становятся функторами, например val output1 = input.map(processor1).map(processor2)

  • маршрутизаторы и фильтры становятся... монадами (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, см. также это и эти статьи.

Другой вариант — использовать Akka как есть, актор гарантирует последовательную обработку, поэтому вы можете реализовать конвейер, создав цепочку акторов:

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)
   }
}

Если вам нужно гарантировать отсутствие гонок между маршрутами, см. этот ответ. Конечно, вы теряете всю идею DSL, используя акторов напрямую.

P.S. Да, вы по-прежнему можете использовать Camel для конечных точек — у Akka есть поддержка для этого. А Акку можно использовать как сервис-активатор.

person dk14    schedule 27.01.2015