1回答
我正在尝试使用akka streams来积累数据并用作批处理： val myFlow: Flow[String, Unit, NotUsed] = Flow[String].collect {如何使用此流等待其他记录到来或30秒空闲？记录一个接一个地来自API请求，我正在尝试在流中积累这些数据，如下所示： Source.single(apiRecord).via(myFlow).runWith(Sink.ignore)
浏览 21提问于2020-09-07得票数 2
回答已采纳
1回答
方法的返回类型是累加器。蓄能器是阿克卡流水池周围的一层薄薄的一层。一个累加器异步地将元素流累加到一个结果中，它可以通过传入一个Akka流源来运行，这将返回一个CompletionStage，当累加器完成时将被赎回。它与Sink>本质上是一样的，实际上它只不过是这种类型的包装器，但最大的区别是，累加器提供了方便的方法，如map、mapFuture、recover等，用于处理结果，就好像它是一个承诺，Sink要求将所有这些操作包装在
浏览 0提问于2017-01-27得票数 1
1回答
我对Akka streams技术非常陌生，我没有找到任何关于我的用例的信息来源。
我希望使用Akka流来使用Kafka主题中的消息，处理它们(同时使用另一个数据源)，并将输出存储在mongo集合中。有可能用阿克卡流吗？谢谢!
浏览 3提问于2022-03-16得票数 1
1回答
我在用可重新启动的源和接收器处理akka流中的错误。newVal + " sink") newVal }}
我是打破源和下沉分别与不同的情况我首先断开接收器，希望接收器重新启动，并一次又一次地重新处理newVal == 3消息。但是似乎接收器中的错误被抛出了，只有源失败被重新尝试，所以源最终被重新启动并重新处理从0
浏览 8提问于2022-01-10得票数 1
回答已采纳
由于SSE支持背压，Akka应该能够做到这一点。我尝试了几种不同的方式，但额外的信息正在被丢弃。
浏览 1提问于2021-08-02得票数 0
我有一个非常简单的流来处理一个文件--只是一个FileIO.FromFile源、一个通过流和一个Last接收器： var result = runnable.Run(Context.Materializer()).Result;
流按预期运行:记录器发送字节串大小，直到文件源完全耗尽为止。，这将关闭源。但是当这个
浏览 0提问于2018-05-17得票数 0
基于：这有什么实际的原因吗？我们能把签名改为def fromIterator(iterator: => Iterator[T])吗？(避免执行Source.fromIterator( () => myIterator) )
浏览 2提问于2016-11-18得票数 5
回答已采纳
下面是一个简单的例子：import akka.stream.ActorMaterializer .to(Sink.ignore)上面的代码段将允许源和接收器在同一个参与者中运行按预期工作。在我被添加一个异步边界之后：
import ak
浏览 3提问于2018-10-19得票数 6
回答已采纳
1回答
Akka Http请求和响应模式
、、、、
我有一个需求，客户端调用通过akka http创建的post端点。一旦请求在post方法中，我就需要将post对象传递给流(由源、多个流和接收器等组成)，并从接收器获取响应，以便可以将响应返回给客户端。我已经阅读了一些文章，看到了下面的代码，但我担心我不想为每个请求实现流。我只想物化一个流，并继续将元素传递给该流。**Source.single(req).runWith(sink)(mat)**
s"
浏览 1提问于2019-11-29得票数 0
从概念上讲，我认为将传入的事件建模为一个单独的流是合适的，而该流碰巧是通过多个TCP连接传递的。这对于我的用例来说很有意义，因为我可以非常简单地将流和汇连接到这个源。import akka.stream.scaladsl._import akka.NotUsed })
浏览 4提问于2016-07-06得票数 5
我正在尝试使用Akka和WebSockets实现服务器推送，并编写了以下程序：
亲切的问候
浏览 0提问于2020-01-15得票数 1
我试图创建变量并将值赋值给一个元素，但在此更改之后，Saxon流无法工作。请建议我如何解决这个问题。<?for-each> </xsl:template></xsl:stylesheet>
模板规则不可流\n*操作数{generation_time} {let $feed_generation_time
浏览 6提问于2022-03-09得票数 0
3回答
我有一个带有单个流/图形的Akka Stream应用程序。我想在源位置测量流量，并每隔5秒记录一次，比如“在最近5秒内收到3条消息”。
浏览 5提问于2017-02-01得票数 4
回答已采纳
2回答
我有一个流，其中我附加了两个S3Sinks。在第一个s3Sink中，我希望文件没有头部，在第二个s3Sink (s3SinkHeaders)中，我希望文件有头部。
浏览 23提问于2020-01-21得票数 0
1回答
用ScalaFX监控Akka流源
、、、、
我试图解决的是以下情况:给定无限运行的Akka流，我希望能够监视流的某些点。我能想到在此时将消息发送到Actor的最佳方式也是Source。这使得我可以灵活地连接单个源或将多个源合并到websocket或任何我想要连接的其他客户端。然而，在这种特殊情况下，我试图将ScalaFX与Akka连接起来，但它并不像预期的那样工作。我的问题是：
import scalafx.In
浏览 1提问于2017-06-10得票数 1
回答已采纳
2回答
我有n源，我想按优先级合并在Akka流中。我的实现基于，其中三个优先级源被合并。我试图用以下方法抽象出Source的数量：import akka.stream.{ClosedShape, Graph, Materializer}
import akka.stream.scaladsl.import akka.stream.流图DSL中可以合并任意数量的Source吗？
浏览 1提问于2018-06-11得票数 0
回答已采纳
我们在集群单例代码中使用来减轻来自读取事件的负载。现在，随着PersistentView不再受欢迎，建议我们使用基于Stream的PersistentQuery。有几种方法可用于将源连接到与先前的PersistentView参与者相对应的参与者：
在PersistentView中，事件是在接收块中处理的，我们有一个基于推送的系统。有了PersistentQuery，对EventsByPersistenceId等的每一个调用都是一次拉力。我如何模仿演员中持续的receive行为？我应该这么做吗？这就是所谓的流<
浏览 3提问于2017-06-22得票数 1
回答已采纳
云服务器
ICP备案
实时音视频
即时通信 IM
云直播
腾讯云开发者
