akka-stream 相关问题

Alpakka Consumer不使用通过Docker编写的Kafka发送的消息

我已经通过Docker compose运行了Kafka和Zookeeper。我可以使用Kafka终端向主题发送/使用消息，并且可以通过Conduktor监视所有内容。但不幸的是，我不是...

如何从Akka Streams Sink抛出的异常中恢复？

我如何从Akka Streams接收器引发的异常中恢复？简单示例：Source integerSource = Source.from（Arrays.asList（1、2、3、4、5、6、7、8、9））; ...

与elasticmq版本不兼容的Akka-http版本

im试图使用elasticmq-rest-sqs运行sqs服务器：“ org.elasticmq” %%“ elasticmq-rest-sqs”％“ 0.14.7”，我的akka 依赖项是：val akka = Seq（“ com.lightbend .akka“ %%” akka-stream -...

如何以惯用的方式向我的Akka流添加错误日志记录？

我当前正在运行类似于以下内容的Akka流设置：┌─────────────┐┐───────────┐│┌ ──────────────┐││REST端点│──▶│队列...

akka.streams。您可以发出值的源（类似于monix.BehaviorSubject）

我正在寻找akka.stream.scaladsl.Source构造方法，该方法可让我简单地从不同代码位置发出下一个值（例如，监视系统事件）。我需要类似于...

akka.UnsupportedAkkaVersion：Akka的当前版本是[2.5.14]，但是akka-http需要版本[2.5.26]

这里是课程：import akka.Done;导入akka.NotUsed;导入akka.actor.ActorSystem;导入akka.http.javadsl.ConnectHttp;导入akka.http.javadsl.Http;导入akka.http.javadsl.model ....

如何减少Akka流中的内存使用量

在scala Play框架中，在操作结果中流式传输ByteArrayOutputStream

我正在尝试在scala播放框架中将svg图像转换为jpeg图像。我用蜡染，它工作正常。现在，我想在操作结果中流式传输输出，而不是转换...

是否可以从Reactor Flux或RxJava2 Observable / Flowable到AkkaStreams源使用适配器？

有时有一些库，例如R2DBC，它返回一个Reactive Stream，即Reactor Flux，但是在Http层中，我们需要另一个Reactive Stream，即AkkaStreams Source，它是AkkaHttp ...

当我引用类型化的actor系统时，如何为AkkaStreams实例化实现器？

下面的代码无法编译，它表示ActorMaterializer缺少隐式的ActorRefFactory。我应该如何提供一个？ val监护人：Behavior [Done] = Behaviors.setup（_ => {Behaviors ....

为什么此Scala Akka代码无法编译？

我遇到了我不理解的编译错误：[错误] /Users/nicu/mooc-reactive/akkahttp-typed-actors-cluster-ORMap-followers/src/main/scala/followers/HttpServer.scala ：43：76：类型...

物化值如何在Akka Stream中工作

我了解到，当我运行Akka流图时，它将实现最正确的组件。但是这样做：Source.range（1,100）.to（Sink.reduce（（a，b）-> a + b））。run（materializer）;将...

Akka：Java中的Flow等效于什么

scala中的Akka流提供了Flow概念。 Java中的等效功能是什么？例如，在Scala中，存在Flow.take（n），但在Java中，存在Source.take（n），它返回的源不是流。

在Akka中是否可以像RxJava / Reactor一样从Source => Source函数创建Flow？

我仍然没有找到在AkkaStreams中进行“过滤”流程的简便方法。使用fromFunction对我来说很容易执行“映射”流程，但是对我却不行。在RxJava / Reactor中有一个...

[使用akka-stream-alpakka从s3下载pdf文件，并将其存储为字节数组

我正在尝试使用akka-stream-alpakka连接器从S3下载pdf文件。我有s3路径，并尝试在alpakka s3Client上使用包装方法下载pdf。 def getSource（s3Path：...

[使用akka-stream-alpakka从s3下载pdf文件

我正在尝试使用akka-stream-alpakka连接器从S3下载pdf文件。我有s3路径，并尝试在alpakka s3Client上使用包装方法下载pdf。 def getSource（s3Path：...

Akka Alpakka SqsSource足够奇怪，可以与queueUrl和queueName一起使用

我正在使用Akka Streams，还使用alpakka.sqs.scaladsl从sqs队列中读取消息。我已经做过很多次了，但是现在我上传了一个版本，该版本将队列名称而不是队列URL放置在源中...

基于第一个输入元素选择一个BidiFlow

我有一个协议，其类型为协议：Flow [ByteString，ByteString，NotUsed]。进入流的元素是用户发送的消息，而离开流的元素是来自...

您如何处理一系列Akka流源？

我们有一个可以处理事件的接收器：def parseEvent（）：Sink [T，Future [akka.Done]] = {Sink.foreach [T] {event => {//处理事件}}}使用单个源即可正常工作：...

将序列输入Akka Stream的Combine（）

您如何使用Akka Stream的Combine方法合并一系列Source？例如，val source = Seq [Source [T，_]]。似乎没有方法签名需要Seq / Iterable / etc ......>