scala - 使用 Akka 流分区时,我得到“已经连接”
问题描述
我正在尝试 Akka Stream API,但我不知道为什么它会抛出 java.lang.IllegalArgumentException。
val graph = RunnableGraph.fromGraph(
GraphDSL.create(source, sink)
((source, sink) => Seq(source, sink)) {
implicit b => (source, sink) =>
Import akka.stream.scaladsl.GraphDSL.Implicits._
val partition = b.add(Partition[(KinesisRecord)](2, flow => {
1
}))
source ~> partition.in
partition.out(0) ~> sink
partition.out(1) ~> sink
ClosedShape
})
这是当前代码。错误如下
[info] - should consume *** FAILED ***
[info] java.lang.IllegalArgumentException: [Map.in] is already connected
[info] at akka.stream.scaladsl.GraphDSL$Builder.addEdge(Graph.scala:1567)
[info] at akka.stream.scaladsl.GraphDSL$Implicits$CombinerBase.$tilde$greater(Graph.scala:1730)
[info] at akka.stream.scaladsl.GraphDSL$Implicits$CombinerBase.$tilde$greater$(Graph.scala:1729)
[info] at akka.stream.scaladsl.GraphDSL$Implicits$PortOpsImpl.$tilde$greater(Graph.scala:1784)
[info] at akka.stream.scaladsl.GraphApply.create(GraphApply.scala:46)
[info] at akka.stream.scaladsl.GraphApply.create$(GraphApply.scala:41)
[info] at akka.stream.scaladsl.GraphDSL$.create(Graph.scala:1529)
我使用 kinesisRecord 作为源的目标。但是,在这段代码中,如果我将 outputPorts 更改为 1 并删除
partition.out(1) ~> sink
这条线,它的工作原理。
我不知道是我遗漏了什么还是只是一个错误。
解决方案
我在我的环境中重现了您的错误并使用该解决方案实施。但我使用的是Int
来源,而不是Kinesis
来源。您可以将其替换为您的数据类型,它可能会起作用。
import akka.actor.ActorSystem
import akka.stream.ClosedShape
import akka.stream.scaladsl.{GraphDSL, Merge, Partition, RunnableGraph, Sink, Source}
import scala.concurrent.duration._
object AkkaStreamWithKinesis extends App {
implicit val system = ActorSystem("AkkaStreamWithKinesisSystem")
val source = Source(1 to 1000).throttle(5, 1 second)
val sink = Sink.foreach[Int](println(_))
val graph = RunnableGraph.fromGraph(
GraphDSL.create(source, sink)
((source, sink) => Seq(source, sink)) {
implicit builder =>
(source, sink) =>
import akka.stream.scaladsl.GraphDSL.Implicits._
val partition = builder.add(Partition[Int](2, flow => {
1
}))
val merge = builder.add(Merge[Int](2))
source ~> partition.in
partition.out(0) ~> merge.in(0)
partition.out(1) ~> merge.in(1)
merge.out ~> sink
ClosedShape
}).run()
}
输出:
09:15:36.443 [AkkaStreamWithKinesisSystem-akka.actor.default-dispatcher-6] INFO akka.event.slf4j.Slf4jLogger - Slf4jLogger started
1
2
3
4
5
6
7
8
9
10
11
12
推荐阅读
- file - .nii 扩展名的文件有分隔符吗?
- mysql - 此查询也显示昨天和今天的数据,但我只想要今天的数据
- jquery - 嵌套 $(document).ready() 以测试嵌套页面是否处于活动状态
- c# - Autofac 无法解析我的注册通用服务
- hashicorp-vault - HarshiCorpe Vault:收集的数据库凭据已过期,这种情况下的最佳做法是什么?
- c# - 使用实体框架异步填充 GridView
- php - strip_tags + html 实体仅获取数字
- c - C:所有字符串文字都有静态存储持续时间吗?
- android - 延迟深度链接到 android 上的外部应用程序
- c# - 如果在数据库中插入新记录,如何更新 SharePoint Online 列表?