akka-stream 相关问题

如何创建方法并在Akka runWith（）中使用此方法

Akka流代码：Source.range（1，100）.map（i-> {return Stream.of（i * 4）.t oList（）;}）。runWith（Sink.foreach（a-> System.out .println（a）），实现器）；我想创建一个方法，说...

如果只有第一个接收器成功，则将元素发送到第二个接收器

我想使用akka流进行跟踪。。我尝试了广播，也尝试了To，但并没有解决我的问题。Source-> DBSink-> APISink。我首先想在交易时将元素发送到数据库...

将元素发送到多个接收器

我想依次使用Akka Stream。源-> Sink1-> Sink2示例：假设我们有List（1,2,3）作为源Elem 1：将其发送到Sink1，如果接收器1成功，将其发送到接收器2

Play 2.5使用的Akka Streams是什么版本？

[从2.5版开始，Play的默认流处理库是Akka Streams。我在任何地方都找不到官方文档，我想知道Play 2.5使用的Akka Stream版本到底是什么？

Java / Kotlin- Akka流Source.reduce如果Source中为null，则不起作用

在Akka Streams Doku中，给出了一个Scala示例，其中包含一个空值的列表。该列表转换为Source，并像示例中那样减少。不幸的是，在Java / Kotlin中，这不是...

在线程“ main”中获取Akka流代码时出错，异常java.lang.NoClassDefFoundError：scala / Function1 $ class

[当我运行akka流代码时，我遇到了以下Java错误：**线程“ main”中的异常java.lang.NoClassDefFoundError：akka.stream.Supervision $$ ...处的scala / Function1 $ class ...

[现在正在使用从我的scala / akka / play应用程序发布的SNS时，获取池正在关闭来自akka的消息

我有一些流大约每分钟醒来，并从数据库中提取一些文档并执行一些操作，最后将消息发送到SNS。滴答间隔当前为每1分钟。 ...

Akka流并行性

根据文档[1]，我一直试图在Akka Stream中并行化流，但是由于某些原因，我没有得到预期的结果。我遵循了文档中概述的步骤，...

将mapAsync提取到AkkaStreams中的方法

我有一个Akka Streams流，有一点我需要对代码块使用mapAsync。 （...）.via（flow1）.mapAsync（1）（response => *几行代码her *）.via（...

是否有可能从Akka Streams Flow中生成多个元素

假设我有一个字符串“ sky is blue”，并且在内部流程中，我将其分为几个元素，我希望将这些元素向下传播。 Source.single（“仅一个元素”）.via（Flow.fromFunction（...

如何持续使用来自Apache Pulsar的消息？

您如何继续使用Akka Streams消耗来自Apache Pulsar的消息并打印每条消息？以下是我从pulsar4s库中找到的示例代码。而不是将消息发布到...

如何使用Akka再次传输文件？

我是Akka的初学者。 （我正在使用Java）我正在使用Akka制作文件传输系统。目前，我已经完成了Actor1（Local）-> Actor2（Remote）文件的发送。现在，当我遇到问题时...

我应该如何测试akka-streams RestartingSource的用法

我正在开发一个具有几个长时间运行的流的应用程序，它在其中订阅有关某个实体的数据并处理该数据。这些流应该达到24/7，所以我们需要...

从akka.stream.scaladsl.Source读取第一个字节

我正在尝试从akka.stream.scaladsl.Source [ByteString，Any]读取前16个字节，并返回[Array [Byte]，Source [ByteString，Any]]。读取前16个字节后，我想流式传输...

Alpakka AMQP：如何检测声明异常？

我有一个带有声明的AMQP源和AMQP接收器：List 声明= new ArrayList （）{{add（QueueDeclaration.create（sourceExchangeName））;添加（...

Java Akka演员和流

我从https://doc.akka.io/docs/akka/current/index.html开始学习akka。 DeviceManager是...

我如何从Akka提取实时FileIO状态？

我正在使用Akka制作文件传输系统。我看文件已经有一段时间了。当前的进度状态是Actor2收到了Actor1发送的文件，并将其写入本地...

如何在Play Framework中正确使用SSE？

我有一个REST服务。当用户发出请求时，服务会向外部服务和数据库发出多个请求，合并结果并回馈响应。像这样的东西：...

物化值是什么意思，为什么在同一图形的以下逻辑中它也不相同？

我是Akka Stream的新手，从上周开始才开始阅读其文档。我能理解大多数概念，但发现很难理解物化价值在...

根据发送的消息恢复Akka的要求

我通过Ask向演员发送不同的消息。超时时，我想提供一个默认值，该默认值对于询问参与者的消息是不同的。由于超时异常始终是...