Akka Stream：mapMaterializedValue 是什么意思

Question

我已阅读Akka streams materialization concept，并了解流物化是：

获取流描述（图表）并分配运行所需的所有必要资源的过程。

我按照一个示例构建了我的 akka 流，使用 mapMaterializedValue 将消息发送到队列。代码的目的是在构建流蓝图并且代码正常工作后将消息推送到队列，但我不太明白 mapMaterializaedValue 在代码中做了什么：

Promise<SourceQueueWithComplete<String>> promise = new Promise.DefaultPromise<>();

Source<String, SourceQueueWithComplete<String>> s = Source
    .queue(100, OverflowStrategy.fail())
    .mapMaterializaedValue(queue -> {
        promise.trySuccess(queue);
    });

source.toMat(Sink.foreach(x -> System.out.println(x)), Keep.left()).run(materIalizer);

promise.<SourceQueueWithComplete<String>>future().map(mapMapperFunction(), actorSystem.dispatcher());

Answer 1

mapMaterializedValue 的目的是在物化后立即转换物化值。例如，假设您有一个接受如下回调的第三方库：

interface Callback<T> {
    void onNext(T next);
    void onError(Throwable t);
    void onComplete();
}

然后您可以创建一个返回 Source<T, Callback<T>> 的方法，您可以在流实际运行时立即将其具体化值传递给该第三方库：

<T> Source<T, Callback<T>> callbackSource() {
    return Source.queue(1024, OverflowStrategy.fail())
        .mapMaterializedValue(queue -> new Callback<T> {
            // an implementation of Callback which pushes the data
            // to the queue
        });
}

Source<Integer, Callback<Integer>> source = callbackSource();

Callback<Integer> callback = source
    .toMat(Sink.foreach(System.out::println), Keep.left())
    .run(materializer);

thirdPartyApiObject.runSomethingWithCallback(callback);

您可以在这里看到，这可以简化必须使用此类第三方 API 的代码，因为您只执行此队列 -> 回调转换一次并将其封装在方法中。

但是，在您的情况下，您并不需要它。您正在使用mapMaterializedValue 完成具有物化值的外部承诺，这是完全没有必要的，因为您可以直接在物化后使用物化值：

Source<String, SourceQueueWithComplete<String>> s = Source
    .queue(100, OverflowStrategy.fail());

SourceQueueWithComplete<String> queue = source
    .toMat(Sink.foreach(x -> System.out.println(x)), Keep.left())
    .run(materIalizer);

mapMapperFunction().apply(queue);