Akka 流 - 根据流中的元素选择接收器

Question

我正在使用 Akka 流创建一个简单的消息传递服务。该服务就像邮件传递一样，其中来自源的元素包括destination和content，例如：

case class Message(destination: String, content: String)

服务应根据destination字段将消息传递到相应的接收器。我创建了一个DeliverySink类来让它有一个名字：

case class DeliverySink(name: String, sink: Sink[String, Future[Done]])

现在，我实例化了两个DeliverySink，让我称它们为sinkX和sinkY，并根据它们的名字创建了一个地图。实际上，我想提供一个接收器名称列表，并且该列表应该是可配置的。

我面临的挑战是如何根据destination字段动态选择合适的接收器。

最终，我想将Flow[Message]映射到水槽。我试过了：

val sinkNames: List[String] = List("sinkX", "sinkY")
val sinkMapping: Map[String, DeliverySink] = 
sinkNames.map { name => name -> DeliverySink(name, ???)}.toMap
Flow[Message].map { msg => msg.content }.to(sinks(msg.destination).sink)

但是，显然这是行不通的，因为我们无法在地图之外引用msg......

我想这不是一个正确的方法。我还考虑过将filter与broadcast一起使用，但是如果目标扩展到 100，我无法键入每个路由。实现目标的正确方法是什么？

[编辑]

理想情况下，我想让目的地充满活力。因此，我无法在过滤器或路由逻辑中静态键入所有目标。如果尚未连接目标接收器，它也应动态创建新的接收器。

Answer 1

如果必须使用多个接收器

Sink.combine将直接满足您现有的要求。 如果您在每个Sink之前附加适当的Flow.filter，则他们只会收到相应的消息。

不要使用多个接收器

总的来说，我认为让流的结构和内容包含业务逻辑是糟糕的设计。 你的流应该是在普通 scala/java 代码中的业务逻辑之上进行背压并发的薄薄外衣。

在这种特殊情况下，我认为最好将目标路由包装在单个 Sink 中，并且逻辑应该在单独的函数中实现。 例如：

val routeMessage : (Message) => Unit = 
(message) => 
if(message.destination equalsIgnoreCase "stdout")
System.out println message.content
else if(message.destination equalsIgnoreCase "stderr")
System.err println message.content
val routeSink : Sink[Message, _] = Sink foreach routeMessage

请注意，现在测试我的routeMessage要容易得多，因为它不在流中：我不需要任何 akka 测试套件"东西"来测试 routeMessage。 如果我的并发设计发生更改，我还可以将函数移动到Future或Thread。

许多目的地

如果您有许多目的地，则可以使用Map。 例如，假设您要将消息发送到 AmazonSQS。您可以定义一个函数来将队列名称转换为队列 URL，并使用该函数维护已创建名称的映射：

type QueueName = String
val nameToRequest : (QueueName) => CreateQueueRequest = ???  //implementation unimportant
type QueueURL = String
val nameToURL : (AmazonSQS) => (QueueName) => QueueURL = {
val nameToURL = mutable.Map.empty[QueueName, QueueURL]
(sqs) => (queueName) => nameToURL.get(queueName) match {
case Some(url) => url
case None => {
sqs.createQueue(nameToRequest(queueName))
val url = sqs.getQueueUrl(queueName).getQueueUrl()
nameToURL put (queueName, url)
url
}
}
}

现在，您可以在单个接收器中使用此非流函数：

val sendMessage : (AmazonSQS) => (Message) => Unit = 
(sqs) => (message) => 
sqs sendMessage {
(new SendMessageRequest())
.withQueueUrl(nameToURL(sqs)(message.destination))
.withMessageBody(message.content)
}
val sqs : AmazonSQS = ???
val messageSink = Sink foreach sendMessage(sqs)

旁注

对于destination您可能想使用除String以外的其他东西。 共积通常更好，因为它们可以与 case 语句一起使用，如果您错过了以下几种可能性之一，您将获得有用的编译器错误：

sealed trait Destination
object Out extends Destination
object Err extends Destination
object SomethingElse extends Destination
case class Message(destination: Destination, content: String)
//This function won't compile because SomethingElse doesn't have a case
val routeMessage : (Message) => Unit = 
(message) => message.destination match {
case Out =>
System.out.println(message.content)
case Err =>
System.err.println(message.content)
}

Answer 2

根据您的要求，也许您想考虑使用 groubBy 将流源多路复用为子流：

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl._
import akka.util.ByteString
import akka.{NotUsed, Done}
import akka.stream.IOResult
import scala.concurrent.Future
import java.nio.file.Paths
import java.nio.file.StandardOpenOption._
implicit val system = ActorSystem("sys")
implicit val materializer = ActorMaterializer()
import system.dispatcher
case class Message(destination: String, content: String)
case class DeliverySink(name: String, sink: Sink[ByteString, Future[IOResult]])
val messageSource: Source[Message, NotUsed] = Source(List(
Message("a", "uuu"), Message("a", "vvv"),
Message("b", "xxx"), Message("b", "yyy"), Message("b", "zzz")
))
val sinkA = DeliverySink("sink-a", FileIO.toPath(
Paths.get("/path/to/sink-a.txt"), options = Set(CREATE, WRITE)
))
val sinkB = DeliverySink("sink-b", FileIO.toPath(
Paths.get("/path/to/sink-b.txt"), options = Set(CREATE, WRITE)
))
val sinkMapping: Map[String, DeliverySink] = Map("a" -> sinkA, "b" -> sinkB)
val totalDests = 2
messageSource.map(m => (m.destination, m)).
groupBy(totalDests, _._1).
fold(("", List.empty[Message])) {
case ((_, list), (dest, msg)) => (dest, msg :: list)
}.
mapAsync(parallelism = totalDests) {
case (dest: String, msgList: List[Message]) =>
Source(msgList.reverse).map(_.content).map(ByteString(_)).
runWith(sinkMapping(dest).sink)
}.
mergeSubstreams.
runWith(Sink.ignore)