gpt4 book ai didi

java - Akka 流 : Simulating a failed stream with OverflowStrategy. 失败()

转载 作者:行者123 更新时间:2023-12-04 02:37:01 24 4
gpt4 key购买 nike

我是 Akka 和 Akka 流的新手。我创建了一个虚拟流,我希望它以异常结束,因为我的 map() 函数非常慢,我将缓冲区设置为 1

所以我的问题分为两部分:

  1. 为什么这段代码可以正常运行?
  2. 如何模拟溢出? (用于学习目的)
    import akka.NotUsed;
import akka.actor.ActorSystem;
import akka.stream.OverflowStrategy;
import akka.stream.javadsl.Sink;
import akka.stream.javadsl.Source;

public class Application {

public static void main(String[] args) {
final ActorSystem system = ActorSystem.create("reactive-test");
Source<Integer, NotUsed> source =
Source.range(0, 10000000)
.buffer(1, OverflowStrategy.fail())
.map(Application::doubleInt);
source.runWith(Sink.foreach(a -> System.out.println(a)), system);
}

private static Integer doubleInt(int i) {
try {
Thread.sleep(2_000);
} catch (Exception e) {
System.out.println(e);
}
return 2 * i;
}
}

最佳答案

Why this code works without a failure?

原因是背压。 Source 不会产生比 Sink 消耗的更多的元素,因此 Slow Sink 直接影响元素产生的速度。因此,您的缓冲区永远不会溢出。

How can I simulate an overflow? (for learning purposes)

有一个渴望消费但同时又很慢的sink。它可以通过添加一个 grouped(1000) 来模拟,该 grouped(1000) 创建 1000 个元素的列表并将其传递到下游。

import akka.NotUsed;
import akka.actor.ActorSystem;
import akka.stream.OverflowStrategy;
import akka.stream.javadsl.Sink;
import akka.stream.javadsl.Source;

public class StreamsBufJava {

public static void main(String[] args) {
final ActorSystem system = ActorSystem.create("reactive-test");
Source<Integer, NotUsed> source =
Source.range(0, 10000000)
.buffer(1, OverflowStrategy.fail())
.grouped(1000)
.mapConcat(list -> list)
.map(StreamsBufJava::doubleInt);


source.runWith(Sink.foreach(System.out::println), system);
}

private static Integer doubleInt(int i) {
try {
Thread.sleep(2_000);
} catch (Exception e) {
System.out.println(e);
}

return 2 * i;
}
}

产生

0
2
[ERROR] [04/17/2020 09:40:47.671] [reactive-test-akka.actor.default-dispatcher-5] [Buffer(akka://reactive-test)] Failing because buffer is full and overflowStrategy is: [Fail] in stream [class akka.stream.impl.fusing.Buffer$$anon$26]
4
6
8

关于java - Akka 流 : Simulating a failed stream with OverflowStrategy. 失败(),我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/61260527/

24 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com