akka-stream

0热度

1回答

使用VisualVM来配置使用akka-streams-kafka的应用程序。它显示了很多卡夫卡协调阻塞线程谁是这些协调员？我也有三个卡夫卡消费者，也阻塞线程我需要为他们创造一个独立的执行上下文？

0热度

1回答

我有一个很重的用户数据流。我想通过它的id来确定这是否是新用户。为了减少对数据库的调用，我宁愿在先前用户的内存中维护一个状态。 val users = mutable.set[String]() //init the state from db user = db.getAllUsersIds() val source: Source[User, NotUsed] val dbSink:

2热度

1回答

Akka Streams Kafka - 消费者的单元测试

我试图通过设置Kafka服务器并使用生产者发送消息来在本地测试我的代码，但我想知道是否有一种方法可以为此编写单元测试一段代码（测试消费者收到的消息是否正确）。 val consumerSettings = ConsumerSettings(system, new ByteArrayDeserializer, new StringDeserializer) .withBootst

0热度

1回答

Akka Streams Hikari连接池用于MySQL流式传输

我使用Slick 3和Akka Streams从mysql中传输数据。这是我建立我的源 import slick.jdbc.MySQLProfile.api._ val enableJdbcStreaming: (java.sql.Statement) => Unit = {statement => if (statement.isWrapperFor(classOf[com.mys

0热度

1回答

如何延迟附加到另一个akka流的akka流的实现？

的文档FlowOps#concat指出 ...源一起使用这个流程化，并从生产，直到它的时间来断言背压元素只是不停。换句话说，如果我有 sourceA.concat(sourceB) 然后sourceB将在同一时间sourceA物化。我如何推迟sourceB正在实现，直到sourceA完成？我正在使用Akka 2.5.6。

1热度

1回答

Akka Streams KillSwitch在alpakka jms

我有一个场景，我开始使用alpakka多个jmsSource（对于不同的队列）。我还需要在任何时候卸下队列。所以我已经添加KillSwitch到jms阿卡流，如下所示： - trait MessageListener { lazy val jmsPipeline = jmsSource .map { x => log.info(s"Received message ${x}

4热度

1回答

Akka流保持空闲，而不是抛出异常

我是Akka/Scala的新手，正在尝试调试下面的代码。当resultSetParser有一个例外时，它不会抛出它。相反，使用此代码的服务只是永远闲置。如何让我的服务抛出异常，而不仅仅是在流中等待？在Akka中有没有类似watchException()的函数，我可以在watchTermination()之后调用它，使它在处理流时看到异常？ val chunkSource: Source[Chun

1热度

1回答

转化油滑流数据，并使用阿卡的Http

目的是从数据库流数据，在这个组块数据的执行一些计算（该计算返回一些情况下类的未来）发送分块响应，并发送该数据作为分块的响应给用户。目前，我能够流式传输数据并发送响应，而无需执行任何计算。但是，我无法执行此计算，然后传输结果。这是我实施的路线。 def streamingDB1 = path("streaming-db1") { get { val src = Source.

0热度

1回答

如何在构建Akka Streams Kafka中的ProducerMessage时配置偏移参数？

我正在使用Scala 2.11和Akka Streams Kafka 0.17。我有一个流其中：甲Source使用Source.actorRef创建。在这里，演员计划以固定间隔运行并持续生成消息，并将消息发送到流中。我已附加Producer作为Flow。制片人推动ProducerMessage.Message成为卡夫卡话题。一些数据库操作。我有一个问题，同时构建ProducerMessa

0热度

1回答

使用akka流时的事件顺序

阅读akka流的文档，我不太清楚诸如消息顺序之类的东西，以及我是否可以执行它。让我用我为聊天服务器编写的一小段代码来设置我的问题的上下文。 def flowShape(user: User) = GraphDSL .create(Source.actorRef[ChatMessage](bufferSize = 5, OverflowStrategy.fail)) { impl