akka 溪流：使用www.example.com()模拟失败的流OverflowStrategy.fail

whhtz7ly 于 2022-11-06 发布在其他

关注(0)|答案(1)|浏览(125)

我是Akka和Akka流的新手。我创建了一个虚拟流，我希望它以一个异常结束，因为我的map（）函数非常慢，我将缓冲区设置为1。
所以我的问题分为两部分：
1.为什么这段代码可以正常工作而不会失败？
1.如何模拟溢出？（用于学习目的）

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;
        }
    }

akka

来源：https://stackoverflow.com/questions/61260527/akka-streams-simulating-a-failed-stream-with-overflowstrategy-fail

1条答案

按热度按时间

zxlwwiss1#

为什么这段代码可以正常工作而不会失败？
原因是反压力。源产生的元素不会比接收器消耗的元素多，因此慢的接收器会直接影响元素产生的速度。因此，缓冲区永远不会溢出。
如何模拟溢出？（用于学习目的）
拥有一个渴望消费但同时又很慢的接收器。可以通过添加一个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

赞(0）回复(0）举报 2022-11-06

我来回答

akka 溪流：使用www.example.com()模拟失败的流OverflowStrategy.fail

1条答案

相关问题

热门标签

最新问答