如何在Akka Stream流运行结束后访问statefulMapConcat的内部状态

方案1：流完成时输出状态元素（兼容性最强） #

Akka Stream 算子内部状态由框架线程模型保证单线程访问，不会存在竞态条件，你可以在流终止时将状态作为特殊元素输出，下游单独捕获即可。
代码示例：

// 先定义元素包装类，区分普通业务元素和最终状态
interface StreamRecord {}
record DataElem(Elem value) implements StreamRecord {}
record StateResult(Set<Elem> state) implements StreamRecord {}

// 改造算子逻辑，Akka 2.6+ 推荐用statefulMap替代statefulMapConcat实现带完成回调的状态管理
Source<StreamRecord, NotUsed> streamWithState = source.via(someBusiness())
  .statefulMap(
      () -> new HashSet<Elem>(), // 初始化状态
      (state, elem) -> {
          state.add(elem); // 原有状态更新逻辑
          return Pair.create(state, new DataElem(elem)); // 输出普通业务元素
      },
      state -> Optional.of(new StateResult(state)) // 流完成时输出最终状态
  );

// 下游分流处理
Pair<Pair<NotUsed, CompletionStage<Void>>, CompletionStage<Set<Elem>>> result = streamWithState
  .alsoToMat(
      // 普通业务元素走原有业务Sink
      Flow.of(StreamRecord.class)
          .filter(elem -> elem instanceof DataElem)
          .map(elem -> ((DataElem)elem).value())
          .to(yourOriginalBusinessSink()),
      Keep.left()
  )
  .toMat(
      // 单独捕获状态元素
      Flow.of(StreamRecord.class)
          .filter(elem -> elem instanceof StateResult)
          .map(elem -> ((StateResult)elem).state())
          .to(Sink.head()),
      Keep.both()
  )
  .run(system);

// 最终状态可以通过CompletionStage异步获取，完全无锁
CompletionStage<Set<Elem>> finalState = result.second();

该方案所有状态操作都在Akka Stream算子的安全上下文中执行，无任何额外锁开销，也不会出现并发访问问题。

方案2：通过CompletableFuture传递状态（无需修改流输出结构） #

如果你不想改动原有流的元素结构，可以在流构造时绑定一个CompletableFuture接收最终状态：

// 每个流实例对应独立的future，避免状态混淆
CompletableFuture<Set<Elem>> stateFuture = new CompletableFuture<>();

source.via(someBusiness())
  .statefulMap(
      () -> new HashSet<Elem>(),
      (state, elem) -> {
          state.add(elem); // 原有状态更新逻辑
          return Pair.create(state, elem); // 输出原业务元素，不改动流结构
      },
      state -> {
          // 流完成时完成future，传递状态
          stateFuture.complete(state);
          return Optional.empty(); // 不输出额外元素
      }
  )
  .to(yourOriginalBusinessSink())
  .run(system);

// 异步获取状态
CompletionStage<Set<Elem>> finalState = stateFuture;

注意每个流运行实例都要创建独立的stateFuture，不要在多个流之间共享，避免状态串扰。

核心原理 #

Akka Stream 每个算子实例的内部状态都由框架保证仅由单个线程访问，不存在并发修改问题，只要不将状态泄露到算子外部的并发场景，完全不需要额外加锁。两种方案都严格遵循Akka Stream的线程安全约定，性能远高于全局并发容器方案。

内容的提问来源于stack exchange，提问作者Sigurd