更改akka streams的源数据

Question

我正在学习JavaAkka流，并使用https://doc.akka.io/docs/akka/current/stream/stream-flows-and-basics.html定义了以下内容：

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;

public class SourceExample {

    static ActorSystem system = ActorSystem.create("SourceExample");

    public static void main(String args[]) throws ExecutionException, InterruptedException {

        final List<Integer> sourceData = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);

        final Source<Integer, NotUsed> source =
                Source.from(sourceData);
        final Sink<Integer, CompletionStage<Integer>> sink =
                Sink.<Integer, Integer>fold(0, (agg, next) -> agg + next);

        final CompletionStage<Integer> sum = source.runWith(sink, system);

        System.out.println(sum.toCompletableFuture().get());
    }

}

运行此代码的行为符合预期。

Akka Streams正在解决这个代码可以重复执行的问题。

在现实场景中，sourceData不会是静态的，Akka Streams对如何处理不断变化的数据有意见吗，还是由开发人员决定？

在最简单的情况下，只需在源数据更改时每X分钟重新执行一次流流（例如使用计划的任务）。还是Akka流是长期存在的，源数据更改并且流计算是根据某些参数重新执行的？

Akka Streams留档定义了多个数据源，但我不明白如何使用Akka Streams来处理不断变化的源数据。

Answer 1

Akka Streams可以而且经常运行到（在不久之前）你的应用停止。例如，流消费（例如，使用来自Alpakka Kafka的Kafka消费源）Kafka记录通常在应用程序中很早就开始，直到应用程序被杀死才停止。

为了详细说明，流一直运行到：

• 阶段信号完成（例如，在您的示例中，Source.from将在发出10后发出完成信号）
• 阶段失败（通常引发异常）

对动态数据有用的示例源代码是source（不引入Alpakka或Akka-HTTP）。队列，具体化为一个队列，队列中的元素可供流使用。