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

Question

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

在一个组件中，我们注意到 Streams 没有处理任何消息，尽管提供的 Publisher 提供了记录。我将问题确定为以下情况：

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();

一个组件提供一个 Publisher（它需要是 Publisher，因为 API 使用 Reactive Streams API，而不是 Akka Streams API）。这个 Publisher 是从另一个 Akka Streams Source 创建的，并使用 Sink.asPublisher.

变成了一个 Publisher

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

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

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

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

有人知道哪里出了问题吗？

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 创建的源之前就丢弃了元素。

文档说它确实不 drop:

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 例中有效。但是以下在所有情况下都有效（我只是添加了一个节流阀来减慢生成元素的速度）：

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

因此在我看来，BroadcastHub 有时会在没有连接 Sink 时丢弃元素。

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

Why is a Publisher created with Sink.asPublisher not working when it is consumed by BroadcastHub?

java

akka-stream