如何访问Akka流的计算结果？

Question

我正在尝试返回一个流操作的结果,在本例中为：

1. 求和列表
2. 取值平方
3. 取值平方

其表示为：

        .fold(0, (aggr, next) -> aggr + next)
        .map(x -> x * x)
        .map(x -> x * x)

若要访问我使用的值，请执行以下操作

final AtomicInteger returnValue = new AtomicInteger();

紧随其后的是：

        .to(Sink.foreach(x -> {
            returnValue.set(x);
            System.out.println("got: " + x);
        }))

这需要一个阻塞调用来允许流完成,这是不可接受的：

Thread.sleep(2000);

如果我使用：

    CompletableFuture<Object> futureValue =
            ask(actorRef, Done.done(), Duration.ofMillis(5000)).toCompletableFuture();
    System.out.println(futureValue.toCompletableFuture().get().toString());

返回错误：

Caused by: akka.pattern.AskTimeoutException: Ask timed out on [Actor[akka://StreamsExamples/system/Materializers/StreamSupervisor-0/$$a-actorRefSource#1663100910]] after [5000 ms]. Message of type [akka.Done$]. A typical reason for `AskTimeoutException` is that the recipient actor didn't send a reply.

在本例中，收件人执行元是源，并在done.done消息中返回以下内容：

return Optional.of(CompletionStrategy.immediately());

Akka流可以用来返回流中的计算值吗？唯一的替代方法是将计算值存储在DB中，或者当在以下位置计算值时将其发送到Kafka主题：

.to(Sink.foreach(x -> {

？

完成SRC：

import akka.Done;
import akka.actor.ActorRef;
import akka.stream.CompletionStrategy;
import akka.stream.OverflowStrategy;
import akka.stream.javadsl.Sink;
import akka.stream.javadsl.Source;

import java.util.Arrays;
import java.util.Optional;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;

public class GetStreamValue {

    final static akka.actor.ActorSystem system = akka.actor.ActorSystem.create("StreamsExamples");

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


        int bufferSize = 100;
        final Source<Integer, ActorRef> source =
                Source.actorRef(
                        elem -> {
                            // complete stream immediately if we send it Done
                            if (elem == Done.done()) {
                                return Optional.of(CompletionStrategy.immediately());
                            }
                            else {
                                return Optional.empty();
                            }
                        },
                        // never fail the stream because of a message
                        elem -> Optional.empty(),
                        bufferSize,
                        OverflowStrategy.dropHead());

        final AtomicInteger returnValue = new AtomicInteger();

        final ActorRef actorRef = source
                .fold(0, (aggr, next) -> aggr + next)
                .map(x -> x * x)
                .map(x -> x * x)
                .to(Sink.foreach(x -> {
                    returnValue.set(x);
                    System.out.println("got: " + x);
                }))
                .run(system);

        Arrays.asList(1, 2, 3).forEach(i -> actorRef.tell(i, ActorRef.noSender()));
        Arrays.asList(1,2).forEach(i -> actorRef.tell(i, ActorRef.noSender()));
        actorRef.tell(Done.done(), ActorRef.noSender());

        Thread.sleep(2000);

        System.out.println("returnValue is "+returnValue);

    }
}

Answer 1

我想你可能缺少的是理解Akka Streams中物化价值的概念。浏览文档的这一部分，特别是关于组合物化值的部分。我也尝试在这里解释这个概念（物化价值搜索）。如果你追求物化的价值，那么也许我在这里写的东西会更有意义。

调用source.actorref(..)返回source ，其中T是流经流的元素的数据类型（在您的示例中是integer)，而actorref是该source的物化值。当对runnablegraph调用run时，您可以同步获得物化值，这是to(...)call返回的值。

actorref是如何按照source.actorref(...)语义“驱动”流的。

现在的问题是如何获得通过流的数据。在您的示例中，将所有整数缩减为一个，因此可以使用sink.head，而不是使用sink.foreach（...）（这对副作用有好处）。您看到，sink还可以生成物化值，对于sink.head来说，它物化为流中第一个元素的completionstage，在您的情况下，该元素是唯一的元素。让我们试试：

final ActorRef actorRef = source
                                .fold(0, (aggr, next) -> aggr + next)
                                .map(x -> x * x)
                                .map(x -> x * x)
                                .to(Sink.head())
                                .run(system);

好吧，那没有太大帮助。您仍然只能获得源的物化值。要获得接收器的物化值，我们需要显式地请求它：

final Pair<ActorRef, CompletionStage<Integer>> matVals =
      source
        .fold(0, (aggr, next) -> aggr + next)
        .map(x -> x * x)
        .map(x -> x * x)
        .toMat(Sink.head(), Keep.both())
        .run(system);

现在，我们得到了源和宿的物化值。您可以像前面一样通过actorref驱动流：

final ActorRef actorRef = matVals.first();

Arrays.asList(1, 2, 3).forEach(i -> actorRef.tell(i, ActorRef.noSender()));
Arrays.asList(1,2).forEach(i -> actorRef.tell(i, ActorRef.noSender()));
actorRef.tell(Done.done(), ActorRef.noSender());

您还可以使用completablestageAPI从流中获取值。像这样说：

Integer folded = matVals.second().toCompletableFuture().join(); 

是的，这是阻塞，但您需要在流运行到完成之前以某种方式阻止主线程完成。