Akka流图恢复问题
是指在使用Akka流进行数据处理时，当系统出现故障或异常情况导致流图中的某个组件无法正常工作时，如何进行恢复和处理的问题。
Akka是一种基于Actor模型的并发编程框架，它提供了一种可靠、高效的消息传递机制，用于构建分布式、可伸缩的应用程序。Akka流是Akka框架中的一个模块，用于处理数据流的传输和转换。
当出现Akka流图恢复问题时，可以采取以下几种方式进行处理：
- 监督和重启：Akka框架提供了监督机制，可以监控流图中的组件，并在组件发生故障时进行重启。通过配置监督策略，可以定义组件的重启行为，例如重启次数、重启间隔等。这样可以保证系统的可靠性和稳定性。
- 错误处理和恢复策略：在Akka流中，可以使用各种操作符和处理器来处理错误和异常情况。例如，可以使用
mapError操作符将错误转换为特定的类型，并使用recover操作符定义错误恢复策略。这样可以在出现错误时，对数据进行适当的处理和恢复。 - 有状态的流处理：Akka流支持有状态的流处理，可以使用
statefulMapConcat、statefulMapAsync等操作符来维护和更新状态。当组件发生故障时，可以通过恢复状态来保证数据的完整性和一致性。 - 监控和日志记录：在Akka流中，可以使用监控工具和日志记录来监测系统的运行状态和异常情况。通过监控和日志记录，可以及时发现问题并进行处理。
- 腾讯云相关产品推荐：腾讯云提供了一系列与云计算和大数据处理相关的产品和服务，可以用于构建和部署Akka流图。例如，腾讯云的云服务器（CVM）可以提供可靠的计算资源，腾讯云数据库（TencentDB）可以提供高性能的数据存储和管理，腾讯云容器服务（TKE）可以提供容器化部署和管理等。具体产品介绍和链接地址如下：
- 腾讯云云服务器（CVM）：提供可靠的计算资源，支持多种规格和配置，适用于各种应用场景。详细介绍请参考：腾讯云云服务器
- 腾讯云数据库（TencentDB）：提供高性能的数据存储和管理服务，支持关系型数据库（MySQL、SQL Server等）和NoSQL数据库（MongoDB、Redis等）。详细介绍请参考：腾讯云数据库
- 腾讯云容器服务（TKE）：提供容器化部署和管理服务，支持Docker容器和Kubernetes集群，提供高可用性和弹性伸缩的容器化解决方案。详细介绍请参考：腾讯云容器服务
通过以上方式和腾讯云相关产品，可以有效地处理和恢复Akka流图中的问题，保证系统的稳定性和可靠性。
相关·内容
1回答
反应性-卡夫卡流消费者:死信发生
、、、、
我试图使用akka的反应性卡夫卡库来消费卡夫卡的信息。我得到了一条信息，然后我得到了
[INFO] [01/24/2017 10:36:52.934] [CommittableSourceConsumerMain-akka.actor.default-dispatcher-5] [akka://CommittableSourceConsumerMain/system/kafka-consumer-1] Message [akka.kafka.KafkaConsumerActor$Internal$Stop$] from Actor[akka://CommittableSourceConsume
浏览 4提问于2017-01-24得票数 3
回答已采纳
1回答
在Akka流应用程序中，如果应用程序崩溃/节点崩溃，我就有丢失正在运行的消息的风险。
我们可以做些什么来确保我们不会丢失消息，并且一旦应用程序回来就会播放它们。
在我的应用程序中，mapAsyns执行IO调用和一些次要的CPU绑定任务。
在纯Akka应用程序中，我将选择持久队列或事件源。但是，如果是Akka流，因为我无法访问底层队列，我如何确保流中没有消息丢失？
浏览 0提问于2016-12-12得票数 1
回答已采纳
1回答
我正在使用SDL框架( OData OData Framework，)和Spring开发一个Java。
当通过Security向应用程序添加安全性时，我面临一些问题。
在一些实现SDL提供的接口的类中，当一个方法被注释为"@PreAuthorize("hasRole('ADMIN')")“时，我会遇到错误。
控制台输出是：
org.springframework.security.authentication.AuthenticationCredentialsNotFoundException: An Authentication object was
浏览 6提问于2016-07-26得票数 0
1回答
我在玩Akka，我试图在物化之后找出它的灵活性。
这样做的一种方法是使用低级的反应性流API：
但是，您需要定义要发布或订阅的这些点。是否有发布或订阅任意物化流图节点的方法？这应该是可能的，因为物化的流图只不过是参与者的集合。
例如:首先，部署流程图1：~> B ~> C
然后，部署流程图2和3: d ~> B ~> E
浏览 4提问于2015-02-17得票数 11
回答已采纳
我已经添加了最新的akka-http到我的项目，但这包括非常老的2.4.19版本的akka-演员。因此，我还在依赖项中添加了akka-参与者版本2.5.4。然而，这会导致以下错误：
Detected java.lang.NoSuchMethodError error, which MAY be caused by incompatible Akka versions on the classpath.
我的maven配置如下：
<dependencies>
<dependency>
<groupId>com.typesafe.akka
浏览 1提问于2017-08-27得票数 2
回答已采纳
我们有以下架构 SQS (源) SQS轮询器->我们的业务逻辑->接收器，它从SQS中删除消息。 这是一个akka流(我们的业务逻辑有多个阶段)。 现在，我们希望通过添加HTTP服务器(而不是Akka HTTP)来扩展此体系结构。 现在我们的服务也有了一条路径 HTTP服务器->我们的业务逻辑->接收器，它完成了一个未来，表明HTTP响应已经完成。 现在，每当HTTP请求到来时，我都需要一种机制来调用流。 现在，SQS源代码本质上是一个长时间运行的线程，它调用服务并将消息推送到akka流的其余部分。 我本质上是在尝试创建一个“可调用的”akka源代码，这样只有当我们收
浏览 14提问于2019-01-09得票数 0
回答已采纳
我正在使用akka创建一个可运行的图形。没有编译时误差，wrt进/出流组件.运行时抛出以下错误：
有什么想法我应该核实一下才能让它运行吗？
requirement failed: The inlets [] and outlets [] must correspond to the inlets [in] and outlets [out]
at scala.Predef$.require(Predef.scala:219)
at akka.stream.Shape.requireSamePortsAs(Shape.scala:168)
at akka.stream.impl.StreamLa
浏览 1提问于2016-02-17得票数 14
回答已采纳
1回答
在写这个问题的时候，我正在使用SpringFramework5.0.0.M5和Akka 2.4.17。在项目中，我目前正在使用来自Akka的参与者消息和流。我明白了，v5也有流。
根据
作为一个反应性引擎/SPI，反应堆核心和IO模块都为集中的用例公开反应流结构，最终与Spring、RxJava、Akka流、Ratpack相结合.作为一个反应性API，反应堆框架模块将提供丰富的易耗特性，如组合和发布子事件。“
它看起来像Spring在引擎盖下使用Akka流(加上其他东西)。
问题是:从Akka流切换到Spring流可能有哪些优点和缺点？
编辑:这里有更广泛的问题(Reactov对Akka一般)
浏览 2提问于2017-02-25得票数 0
回答已采纳
1回答
我使用恢复方法捕捉Akka流中的错误或异常。它是在线性图上工作，而不是非线性图(例如广播，Zip)。为什么图形与扇入或扇出等待失败端口永远，所以Akka流挂断。该解决方案将在的第9节中进行描述。
该帖子使用Try monad并捕获流中的异常。这是可行的。但是，我使用恢复方法，因为我有很多流，我想在一个地方捕获错误。
我准备了下面的例子，但不工作..。
Source(1 to 10)
.via(graph)
.withAttributes(ActorAttributes.supervisionStrategy(Supervision.resumingDecider))
.runFor
浏览 0提问于2018-12-12得票数 1
回答已采纳
我是Akka Stream的新手，从上周起我才开始阅读它的文档。我能够理解大多数的概念，但我发现很难理解Materialized Value在Akka流中的含义及其意义是什么？
如果有人能用一个真实世界的例子来向我解释这一点，那将大大有助于将它与Akka Stream中的用例关联起来。
更新
我正在看下面的例子，想要理解对于同一张图我们得到的两种不同的输出。我知道它和Materialized Value有直接的关系，所以我问了上面的问题。
implicit val system = ActorSystem("PlainSinkProducerMain")
implici
浏览 5提问于2017-08-03得票数 9
回答已采纳
在探索Akka流时，我还遇到了Apache哪个流处理引擎。Akka流实现反应流并支持背压。
那么，如果我要在两者之间做出决定，我应该选择哪一个呢？它们有何不同之处，相似之处何在？这里的标准应该是什么？
浏览 3提问于2020-04-23得票数 4
回答已采纳
我必须实现一个进程，它从队列中轮询消息并将数据转发到http端点。通常，我认为最好的方法是使用Akka流(带有akka http)，因为它为您处理背压等。问题是，我必须在运行时(基于http调用)创建多个不同配置的管道。另一个要求是，应该可以在运行时动态地停止其中的一个管道，然后用另一个配置启动它。
目前，我不确定是否真的能够在运行时动态地旋转新的平行akka流管道(并删除和重新启动其中的一些管道)。对我来说，现在使用akka演员做这个听起来更好(基于队列使用者参与者和akka http参与者，我和我可以在运行时动态地创建一对)，但是要使用它们，我还必须手动实现背压等等。
你认为可以通过akk
浏览 1提问于2018-07-02得票数 0
回答已采纳
2回答
我有一个关于使用Akka Streams和Akka集群的问题。我正在尝试做一个使用Akka Streams和Akka Cluster的分布式字数统计版本。
我想构建一个Akka Streams客户端，它将文本文件作为流I/O读取，并将单词流发送到远程集群。这是客户端的代码：
final Path file = Paths.get("example.txt");
final Source<ByteString, CompletionStage<IOResult>> read = FileIO.fromPath(file);
final Source<
浏览 5提问于2018-11-10得票数 2
1回答
我是非常新的akka-http，我想流一个csv与任意数量的行。
例如，我想返回：
a,1
b,2
c,3
使用以下代码
implicit val actorSystem = ActorSystem("system")
implicit val actorMaterializer = ActorMaterializer()
val map = new mutable.HashMap[String, Int]()
map.put("a", 1)
map.put("b", 2)
map.put("c", 3)
val `text/c
浏览 2提问于2016-10-27得票数 1
我想做以下使用akka streams..我尝试广播，和alsoTo，但它没有解决我的问题
Source -> DBSink -> APISink。
我首先想把元素发送到DB，当事务成功完成后，我想把它发送到APISink..如果first Sink中的事务失败，我不想将其发送到API。
浏览 0提问于2020-06-11得票数 0
2回答
我曾与TPL数据流合作过。真的很喜欢。我从我的java/scala朋友那里听说过很多次Akka这个词，所以我试着阅读它，发现akka也有一个.net端口。太棒了。当我继续阅读akka是什么时，我惊讶地发现它听起来完全像TPL数据流。
那么在我的问题上，TPL数据流和Akka.net有什么区别呢？
什么时候选择什么？
浏览 2提问于2016-09-01得票数 10
回答已采纳
2回答
我开始学习，它是一个处理具有背压功能的数据的框架。库是的一部分，它将自己描述为：
Akka是一个工具包和运行时，用于在JVM上构建高度并发、分布式和弹性强的消息驱动应用程序。
这些能力来自阿克卡演员的本性。然而，在我看来，流处理和参与者是无关的概念。
问题: Akka Streams是否利用了Akka演员的这些特点？如果是，你能解释演员如何帮助溪流吗？
浏览 2提问于2015-07-31得票数 4
回答已采纳
我正在使用Akka.Net (v1.3.2)，并试图使用特定的标记查询所有事件的事件日志。我只想要查询日志时存在的事件。在演员内部，我有以下代码：
var readJournal = PersistenceQuery.Get(Context.System).ReadJournalFor<SqlReadJournal>(SqlReadJournal.Identifier);
var stream = readJournal.CurrentEventsByTag("The Tag Name", Offset.NoOffset());
var materializer =
浏览 2提问于2018-05-07得票数 1
回答已采纳
我想知道，是否可以将代理/参与者库(Akka、轨道、类星体、JADE、Reactors.io)应用于服务环境(OpenWhisk、AWS )？
说得通吗？
如果是的话，什么是最小的示例hat显示了附加值(当我们只使用FaaS或只使用参与者/代理库时，这是缺失的)？
如果不是，那么我们是否能够构造决策图，这可以帮助我们决定，对于我们的问题，我们是否应该使用参与者/代理库或FaaS (或其他什么)？
浏览 6提问于2017-10-22得票数 0
回答已采纳
我正在尝试在akka-http websocket服务器和简单的javascript之间建立一个基本的连接。
在大约20个连接中，有1个成功，其余的失败。我不知道为什么连接的设置如此不可靠。
Application.scala：
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.stream.ActorMaterializer
import services.WebService
import scala.concurrent.Await
import scala.concurrent.durat
浏览 0提问于2015-07-23得票数 2
这是我之前的问题：
我已经成功地使用FileIO.fromPath(Paths.get(file.toURI()在Akka流上发送了文件，它工作得很好。但是，我想在发送文件之前对其进行压缩和加密。我已经创建了方法，打开FileInputStream，通过压缩流路由它，然后通过加密流路由它，现在我想使用Akka流将它定向到套接字中。
文件-> FileInputStream -> CompressedInputStream -> EncryptedInputStream -?> Spring/Akka流
关键是，我可以实时地压缩/加密文件，同时逐块读取它(我没有在磁盘上创
浏览 2提问于2017-02-25得票数 1
回答已采纳
我有下面的示例参与者代码。
object SomeExternalDep {
private var flag = true
// this function will throw an exception once when called with the value 3, then it won't throw another exception
@throws[RuntimeException]
def potentiallyThrows(curr: Int): Unit = {
if (curr == 3 && flag) {
浏览 0提问于2019-07-10得票数 1
回答已采纳
我正在使用Akka，我希望在运行时构建一个定制的RunnableGraph，这取决于一些事先不知道的配置。
例如：
RunnableGraph.fromGraph(GraphDSL.create() { implicit builder =>
val source = builder.add(mySource)
if (A) {
builder.add(flowX)
} else {
builder.add(flowY)
}
ClosedShape
})
这是可能的吗，还是有一种更优雅的方法在运行时组成一个可运行的图，然后再传递和执行？
浏览 8提问于2022-07-27得票数 1
回答已采纳
1回答
如何从Akka做SQL？
、、、、
从Akka应用程序发出SQL语句的惯用Akka方法是什么？
具体来说，我有一个Akka Http REST端点，它希望执行SQL命令。是否有正式的SQL支持或异步或消息传递样式的SQL库？
浏览 5提问于2017-02-22得票数 2
回答已采纳
我正在使用Akka流从AWS S3桶读取.gz文件，应用程序读取数据块并对其进行处理。
当我从一开始读取文件时，下面的代码运行良好，但是如果我从特定的偏移量启动文件，则解压缩失败并抛出ParsingException。
代码：
object StreamApp extends App {
// Run 1
//startStreaming(0)
// Run 2
startStreaming(33339)
def startStreaming(pointer: Long): Unit = {
println("Stream A
浏览 0提问于2018-02-26得票数 1
1回答
我已经构建了一个定义流的akka图。我的目标是重新格式化我未来的响应并将其保存到一个文件中。这个流程可以概括如下：
val g = RunnableGraph.fromGraph(GraphDSL.create() { implicit builder: GraphDSL.Builder[NotUsed] =>
import GraphDSL.Implicits._
val balancer = builder.add(Balance[(HttpRequest, String)](6, waitForAllDownstreams = false))
v
浏览 0提问于2017-02-06得票数 10
回答已采纳
2回答
我正在试图找出如何从scala流图中获得实际结果。
我在用"com.typesafe.akka" %% "akka-stream-experimental" % "1.0"。
我看了看，但找不到任何例子。
所以，假设我有一个密码
val g = FlowGraph.closed() { implicit builder=>
import FlowGraph.Implicits._
val in = Source.apply(1 until 10)
val plus = Flow[Int].map(_ + 10)
val o
浏览 1提问于2015-09-15得票数 1
回答已采纳
我正在寻找一个解释的行为，我看到了以下代码。当
conn.handleWith被注释掉，我使用netcat建立的TCP客户端连接连接，并在几秒钟内显示由对等服务器断开连接(即服务器断开连接)。当conn.handleWith出现在代码中时，我不会看到断开连接。我最初认为这与为服务器设置的idletimeout有关，但事实并非如此。
那么，当没有处理连接的流时，服务器为什么要断开客户端的连接呢？
package com.example;
import java.util.concurrent.CompletionStage;
import akka.D
浏览 3提问于2021-08-09得票数 0
回答已采纳
我正在尝试将集成到一个严重依赖的应用程序中。该应用程序还利用了。
我在运行时遇到了依赖问题。特别是，当我尝试使用Mongo连接器提供的NoClassDefFoundError方法时，我得到了一个成功/失败包装器的MongoSink.insertOne。一个完整的堆栈跟踪：
java.lang.NoClassDefFoundError: com/mongodb/reactivestreams/client/Success
at akka.stream.alpakka.mongodb.scaladsl.MongoFlow$.$anonfun$insertOne$1(MongoFlow.sc
浏览 0提问于2020-12-30得票数 0
回答已采纳
1回答
我目前正在使用Akka-HTTP构建一个客户端WebSockets消费者。我不想尝试在接收器中进行解析，而是希望将代码包装在一个函数中，该函数发出来自接收器的结果，然后使用该输出(来自该函数)进行进一步处理(更多parsing...etc。)。 我现在可以打印来自Sink的每条消息；但是，函数的返回类型仍然是Unit。我的目标是为落入接收器的每个项目从函数发出一个String，然后使用返回的字符串进行进一步的解析。我已经有了到目前为止的代码(注意:它主要是样板代码)。 import java.util.concurrent.atomic.AtomicInteger
import akka.
浏览 11提问于2020-07-08得票数 0
回答已采纳
2回答
我是斯卡拉和阿克卡的新手。我已经创建了一个新项目，下面是我的build.sbt文件。
name := "akka_essentials"
version := "0.1"
scalaVersion := "2.13.4"
val akkaVersion = "2.5.13"
libraryDependencies ++= Seq(
"com.typesafe.akka" %% "akka-actor" % akkaVersion,
"com.typesafe.akka"
浏览 3提问于2020-11-26得票数 1
回答已采纳
1回答
将Akka流连接到JMS
、、、
我正试图通过Akka流连接到通用消息队列(由软件AG)。我看过关于Camel集成的文档，但是我很难理解组件是如何组合在一起的。例如，我是否必须使用ActiveMQ作为代理？
我以前已经通过MQTT (和星火的MQTTUtils)建立了一个连接，但是由于我想尝试Akka，所以我不认为通过TCP进行MQTT是必要的。[建议}()使用JMS而不是另一种协议，特别是使用第三方工具。因此，我的问题是如何通过JMS正确地设置Akka流到UM。
浏览 4提问于2016-10-25得票数 4
1回答
我有以下scala代码：
import akka.Done
import akka.actor.ActorSystem
import akka.kafka.ConsumerMessage.CommittableOffsetBatch
import akka.kafka.scaladsl.Consumer
import akka.kafka.{ConsumerSettings, Subscriptions}
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Sink
import org.apache.kafka.c
浏览 3提问于2017-11-03得票数 1
回答已采纳
我有一个使用Akka流执行管道的代码。
我的问题是，什么是扩大规模的最佳方式？它也可以使用Akka流来完成吗？
或者它需要转换成演员/其他方式？
代码片段是：
val future = SqsSource(sqsEndpoint)(awsSqsClient)
.takeWhile(_=>true)
.map { m: Message =>
(m, Ack())
}.runWith(SqsAckSink(sqsEndpoint)(awsSqsClient))
浏览 0提问于2017-08-21得票数 0
回答已采纳
1回答
目标：
我希望看到Akka的文件IO尽可能的异步和非阻塞。
我目前的知识状况：
在中，您可以阅读以下内容：
备注
由于当前版本的Akka (2.3.x)需要支持JDK6，所以当前提供的文件IO实现无法使用异步文件IO操作，因为这些操作是在JDK7 (和更新版本)中引入的。一旦Akka可以自由地要求JDK8 (来自2.4.x)，这些实现将被更新以使用新的NIO (即AsynchronousFileChannel)。
目前的akka版本是'2.5.4‘。目前版本的akka流是'2.11‘或'2.12’。在中，省略了abhove的注释，它只是明确地提到，文件IO意味着
浏览 3提问于2017-08-18得票数 0
回答已采纳
编辑：问题已经解决，Artur提供的解决方案最终被添加到编辑中。
我试图实现的想法是，一个TCP服务器允许n个连接，如果我得到一个n+1连接，则不允许连接。
因此，我需要以某种方式取消连接，然后继续将特定的流连接到Sink.cancelled()。
我所拥有的是一个连接到自定义流的IncomingConnection，该流根据连接计数对IncomingConnection进行分区。一旦最大连接计数被破坏，分区逻辑就会将其定向到连接到Sink.cancelled的出口。
期望立即取消连接，但它允许客户端连接，然后在一段时间后断开连接。
也许我遇到了与的答案中提到的相同的问题，其中找不到要处理的流
浏览 2提问于2021-08-10得票数 0
回答已采纳
1回答
使用下面的代码，我尝试使用一个参与者作为源，并发送双精度类型的消息以通过滑动窗口进行处理。
滑动窗口被定义为sliding(2, 2)，以计算发送的每个TWP值序列。
发送消息：
actorRef.tell(10, ActorRef.noSender());
actorRef.tell(20, ActorRef.noSender());
actorRef.tell(30, ActorRef.noSender());
actorRef.tell(40, ActorRef.noSender());
actorRef.t
浏览 29提问于2021-06-14得票数 0
回答已采纳
我正试图向akka-http发送200k信息。
protected val someRouts: Route = pathPrefix("foo") {
pathEndOrSingleSlash {
put {
entity(as[Foo]) { foo =>
log.debug(s"/foo update $foo")
complete(Future(Foo("a")).map(f => s"Got - $f
浏览 5提问于2017-09-05得票数 1
回答已采纳
1回答
我试图返回流操作的结果，在本例中是：
与列表之和
平方值
平方值
其表现为：
.fold(0, (aggr, next) -> aggr + next)
.map(x -> x * x)
.map(x -> x * x)
访问我使用的值
final AtomicInteger returnValue = new AtomicInteger();
其次是：
.to(Sink.foreach(x -> {
returnValue.set(x);
浏览 17提问于2021-04-12得票数 0
回答已采纳
Akka-http 说：
除了将服务器端的套接字绑定视为SourceIncomingConnection之外，还将每个连接视为带有SinkHttpResponse的SourceHttpRequest。
假设我们获得了包含来自多个SourceIncomingConnection的传入连接的合并源。
然后，假设我们从Source获得SourceHttpRequest。
然后，没有问题，我们可以提供一个流来将HttpRequest转换为HttpResponse。
问题就在这里--我们如何才能恰当地减少反应呢？我们如何加入对连接的响应？
用例背后的整个想法是优先处理来自不同连接的传入请求的可能
浏览 5提问于2015-12-16得票数 4
我有一个kafka消费者，如下所示： import akka.actor.ActorSystem
import akka.kafka.scaladsl.Consumer
import akka.kafka.{ConsumerSettings, Subscriptions}
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Sink
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.seria
浏览 64提问于2019-04-21得票数 1
1回答
我正在调用一个使用via的方法，如下所示：
myRawStr(id)
.take(1)
.via(myMethod("someString", someSource)
.zip(Source.fromIterator(() => Iterator.from(1)))
.collect {
...
}
myMethod返回Flow[ByteString, MyValidated[MyClass], NotUsed]类型，但现在它将返回Future[Flow[ByteString, MyValidated[MyClass], NotUsed]] (Note：Future[Fl
浏览 0提问于2018-12-16得票数 0
回答已采纳
我有一个流代码
val source = Source.actorRef[Msg[T]](1000, OverflowStrategy.dropHead)
val flow=source.map(println(_)).to(Sink.onComplete{case _=>println("complete")}).run()
在运行代码之后，我从日志文件中发现，akka创建了两个参与者：
akka://xxxx/flow-0-0-unknown-operation#-969158424]
akka://xxxx/flow-0-1-actorRefSour
浏览 3提问于2016-05-01得票数 2
回答已采纳
1回答
更改akka流的源数据
、、、
我正在学习Java流，并使用定义了以下内容：
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
public class SourceExample {
static ActorSystem system = ActorSystem.create("SourceExample");
public static void main
浏览 1提问于2021-04-10得票数 0
回答已采纳
1回答
我有一个处理一些消息的akka流。当一个事件发生时，流应该创建一个不同akka流的新实例。
目前这就是我正在做的事情。这是最好的方法吗？
if(event.happened) new AnalysisFlow(info.id,info.time).flow
谢谢
浏览 0提问于2016-04-24得票数 0
2回答
我有一个Actor，它被设计为与akka-io acking一起工作，这样当它向上游(向网络)发送消息时，它将等待Ack。此参与者是后端异步应用程序的接口。
我希望有一个包装层，它允许我将这个Actor转换成一个akka流Flow[Incoming, Outgoing, ???]，这样它就可以与预期有这样一个签名的新库集成起来。
(来自上游的消息很少，所以我们不太在意向后施压，但拥有它并不是一件坏事。)
sealed trait Incoming //... with implementations
sealed trait Outgoing //... with implementation
浏览 3提问于2015-05-27得票数 0
1回答
我要在日志里找个死信演员。伐木是水平信息。但我不知道sActor的哪种方法会给出一纸空文。下面是日志消息
akka.actor.DeadLetterActorRef - Message [scala.util.Right] from Actor[akka://marksun-app/user/marksun-api/sActor/$a#-396914220] to Actor[akka://marksun-app/deadLetters] was not delivered. [2] dead letters encountered. This logging can be turned o
浏览 4提问于2021-09-18得票数 2
我有一个带有Akka流的管道，它使用自定义调度程序。
Akka流代码：
// Pulls in data via network requests
source
// Downstream will be sent to a different actor.
// Also, a "buffer" of 16 elements will be created between those actors.
.async
.via(writeToDatabase())
.toMat(Sink.ignore)(Keep.right)
.withAttribute
浏览 15提问于2021-09-03得票数 1
3回答
Akka流在java中按id排序
、、、、
我需要在java中按id对我的akka流列表进行排序我在akka source中有对象列表：
SystemCodeTracking{id=9, EntityName='table3'}
SystemCodeTracking{id=2, EntityName='table2'}
SystemCodeTracking{id=10, EntityName='table1'}
我需要将其排序为：
SystemCodeTracking{id=2, EntityName='table2'}
SystemCodeTracking{id=9, E
浏览 2提问于2017-06-20得票数 0
1回答
嗨，我和Kafka和Akka Streams一起工作。在Kafka中，对于主题MyTestTopic，我有3个分区，数据在高度并发的情况下被推入主题，大致上是1000 QPS，它只会比这个更高。
下面是我为Akka Stream Kafka使用者编写的代码：
final ConsumerSettings<String, byte[]> consumerSettings =
ConsumerSettings.create(kafkaConfig, new StringDeserializer(), new ByteArrayDeserializer())
浏览 5提问于2019-12-09得票数 0
云服务器
ICP备案
对象存储
腾讯会议
实时音视频
腾讯云开发者
