java - 为什么使用 Sink.asPublisher 创建的 Publisher 在被 BroadcastHub 使用时不起作用？

Question

我们有一个多组件应用程序，它在组件之间提供 Reactive Streams API。一些组件是使用 Akka Streams 实现的，其他组件则使用例如 react 堆。

在一个组件中，我们注意到 Streams 没有处理任何消息，尽管提供的发布者提供了记录。我将问题归结为以下情况:

Publisher<String> stringPublisher = Source
    .from(Lists.newArrayList("Hello", "World", "!"))
    .runWith(Sink.asPublisher(AsPublisher.WITH_FANOUT), materializer);

Source<String, NotUsed> allMessages = Source
    .fromPublisher(stringPublisher)
    .toMat(BroadcastHub.of(String.class, 256), Keep.right())
    .run(materializer);

allMessages
    .runForeach(System.out::println, materializer)
    .toCompletableFuture()
    .get();

一个组件提供了一个发布者(它需要是发布者，因为 API 使用 Reactive Streams API，而不是 Akka Streams API)。此发布者是从另一个 Akka Streams 源创建的，并使用 Sink.asPublisher 将其转变为发布者。

当我们现在使用 BroadcastHub 从发布者开始具体化流时，根本不会处理任何记录。

我对 Reactor Publisher 进行了同样的尝试:

Publisher<String> stringPublisher = Flux.fromIterable(Sink.asPublisher(AsPublisher.WITH_FANOUT), materializer);

这按预期工作。不幸的是，我不能排除另一个组件从 Akka Stream Source 创建其发布者的情况。

有人知道出了什么问题吗？

Answer 1

我现在知道如何解决它，如果我开始在mapMaterializedValue中使用BroadcastHub的结果源，它就会起作用:

Publisher<String> stringPublisher = Source
    .from(Lists.newArrayList("Hello", "World", "!"))
    .runWith(Sink.asPublisher(AsPublisher.WITH_FANOUT), materializer);

Source
    .fromPublisher(stringPublisher)
    .alsoToMat(BroadcastHub.of(String.class, 256), Keep.right())
    .mapMaterializedValue(source -> source
         .runWith(Sink.foreach(System.out::println, materializer))
    .run(materializer)
    .toCompletableFuture()
    .get();

编辑:TL;DR:Lightbend Forum 中有说明。 :

What happens here is that the main stream is already completed when you attach the other stream. Sometimes it might be quick enough to see a few elements before completion.

---

因此，看起来 BroadcastHub 实际上在消费者附加到 BroadcastHub 创建的 Source 之前删除了元素。

文档说它不会丢弃:

If there are no subscribers attached to this hub then it will not drop any elements but instead backpressure the upstream producer until subscribers arrive.

https://doc.akka.io/docs/akka/current/stream/stream-dynamic.html

实际上，大多数情况下都是如此，但我发现在某些情况下它的行为不正确:

public void testBH3() throws ExecutionException, InterruptedException {
    Publisher<String> stringPublisher = Source
        .from(Lists.newArrayList("Hello", "World", "!"))
        .runWith(Sink.asPublisher(AsPublisher.WITH_FANOUT), materializer);

    Source<String, NotUsed> allMessages = Source
        .fromPublisher(stringPublisher)
        .toMat(BroadcastHub.of(String.class, 256), Keep.right())
        .run(materializer);

    allMessages
        .runForeach(System.out::println, materializer)
        .toCompletableFuture()
        .get();
}

public void repeat() throws ExecutionException, InterruptedException {
    for (int i = 0; i < 100; i++) {
        testBH3();
        System.out.println("------");
    }
}

这对 100 种情况中的 3 种有效。但以下方法在所有情况下都有效(我只是添加了一个 throttle 以减慢生成元素的速度):

public void testBH3() throws ExecutionException, InterruptedException {
    Publisher<String> stringPublisher = Source
        .from(Lists.newArrayList("Hello", "World", "!"))
        .runWith(Sink.asPublisher(AsPublisher.WITH_FANOUT), materializer);

    Source<String, NotUsed> allMessages = Source
        .fromPublisher(stringPublisher)
        .throttle(1, Duration.ofSeconds(1))
        .toMat(BroadcastHub.of(String.class, 256), Keep.right())
        .run(materializer);

    allMessages
        .runForeach(System.out::println, materializer)
        .toCompletableFuture()
        .get();
}

因此，在我看来，当尚未连接任何 Sink 时，BroadcastHub 有时会丢弃元素。