Akka Streams：向 QueueSource 提供 StreamDetachedException

Question

我们有一个

DownloadFileFlow

类，它使用

Akka Streams

和

Akka Http

从不同域下载文件；有时（~50%）在向

QueueSource

提供时，我们会立即遇到

StreamDetachedException

错误：

java.util.concurrent.CompletionException: akka.stream.StreamDetachedException: Stage with GraphStageLogic akka.stream.impl.QueueSource$$anon$1-queueSource stopped before async invocation was processed
   at java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:332)
   at java.base/java.util.concurrent.CompletableFuture.uniAcceptNow(CompletableFuture.java:747)
   at java.base/java.util.concurrent.CompletableFuture.uniAcceptStage(CompletableFuture.java:735)
   at java.base/java.util.concurrent.CompletableFuture.thenAcceptAsync(CompletableFuture.java:2186)
   at scala.concurrent.java8.FuturesConvertersImpl$CF.thenAccept(FutureConvertersImpl.scala:29)
   at scala.concurrent.java8.FuturesConvertersImpl$CF.thenAccept(FutureConvertersImpl.scala:18)
   at x.y.z.DownloadFileFlow.offer(DownloadFileFlow.java:60)

我们不知道错误的原因是什么，但是

postStop

中有一个

akka.stream.impl.QueueSource

方法，我们对此表示怀疑，但不知道停止的原因：

override def postStop(): Unit = {
  val exception = new StreamDetachedException()
  completion.tryFailure(exception)
}

1. SourceQueue参考：

AtomicReference<SourceQueueWithComplete<FileDownloadEnvelope>> queueRef = new AtomicReference<>()

2. 定义：

RestartSource.onFailuresWithBackoff(
     RestartSettings.create(props.getRecoverMinBackOff(), props.getRecoverMaxBackOff(), RANDOM_FACTOR),
        () -> Source.<FileDownloadEnvelope>queue(props.getBufferSize(), OverflowStrategy.backpressure(), props.getMaxConcurrentOffers())
            .filter(this::checkExpiration)
            .map(x -> Pair.create(createContext(x), x))
            .flatMapConcat(this::authIfNeeded)
            .mapAsyncUnordered(props.getParallelism(), this::process)
            .map(this::reply)
            .mapMaterializedValue(x -> {
                this.queueRef.set(x);
                return x;
            }))
    .runWith(Sink.ignore(), actorSystem);

3. 访问（

   offer

   方法）：

public void offer(FileDownloadEnvelope envelope) {
    queueRef.get().offer(envelope)
        .thenAccept(x -> {
            if (!x.isEnqueued())
                replyError(envelope);
        }).exceptionally(e -> { // StreamDetachedException catched here
            replyError(envelope);
            return null;
        });
}

Akka Http 道具：

akka.http.host-connection-pool.min-connections: "10"
akka.http.host-connection-pool.max-connections: "2000"
akka.http.host-connection-pool.max-open-requests: "4096"
akka.http.host-connection-pool.max-retries: "0"
akka.http.host-connection-pool.client.connecting-timeout: 2s
akka.http.host-connection-pool.client.idle-timeout: 5s

直播道具：

BUFFER-SIZE: "2000"
MAX-CONCURRENT-OFFERS: "2000"
PARALLELISM: "70"
RECOVER-MIN-BACK-OFF: 100ms
RECOVER-MAX-BACK-OFF: 500ms

我们正在使用

akka & akka-stream: 2.6.19

和

akka-http: 10.2.0