java-8 - 为什么 Play 2.5 Akka block 响应一次全部加载

Question

我正在尝试使用带有 Akka 的 PLay 2 在 webapp 中实现 block 响应。然而，所有的响应都是一次来的，而不是逐 block 加载响应。下面是我在 Controller 中创建 block 的代码:

/**
 * 
 */    
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;

import com.google.inject.Inject;
import com.google.inject.Singleton;
import akka.stream.javadsl.Source;
import akka.util.ByteString;
import org.pmw.tinylog.Logger;
import play.cache.CacheApi;
import play.cache.Cached;
import play.filters.csrf.AddCSRFToken;
import play.filters.csrf.CSRF;
import play.libs.Json;
import play.libs.concurrent.HttpExecutionContext;
import play.mvc.Controller;
import play.mvc.Http;
import play.mvc.Http.Cookie;
import play.mvc.Result;

import akka.NotUsed;
import akka.actor.Status;
import akka.stream.OverflowStrategy;
import akka.stream.javadsl.Source;
import akka.util.ByteString;

/**
 * @author Abhinabyte
 *
 */
@Singleton
@AddCSRFToken
public class GetHandler extends Controller {

    @Inject
    private CacheApi cache;

    @Inject
    private HttpExecutionContext httpExecutionContext;

    public CompletionStage<Result> index() {

return CompletableFuture.supplyAsync( () ->
            Source.<ByteString>actorRef(256, OverflowStrategy.dropNew())
                    .mapMaterializedValue(sourceActor -> {

                        CompletableFuture.runAsync(() -> {
                            sourceActor.tell(ByteString.fromString("1"), null);
                            sourceActor.tell(ByteString.fromString("2"), null);
                            sourceActor.tell(ByteString.fromString("3"), null);
                            try {
                                Thread.sleep(3000);//intentional delay
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                            sourceActor.tell(ByteString.fromString("444444444444444444444444444444444444444444444444444444444444444444444444"), null);
                            sourceActor.tell(new Status.Success(NotUsed.getInstance()), null);
                        });

                        return sourceActor;
                    })
    ).thenApplyAsync( chunks -> ok().chunked(chunks).as("text/html"));  

    }

}

以下是 application.conf 中的 Akka 线程池配置:

akka {
  jvm-exit-on-fatal-error = on
  actor {
    default-dispatcher {
      fork-join-executor {
        parallelism-factor = 1.0
        parallelism-max = 64
        task-peeking-mode = LIFO
      }
    }
  }
}

play.server.netty {
  eventLoopThreads = 0
  maxInitialLineLength = 4096
  log.wire = false
  transport = "native"
}

正如您在发送最后一个到最后一个 block 之前看到的那样，我故意延迟响应时间。所以从逻辑上讲，它之前的所有分 block 数据都应该在它之前交付。
但是，就我而言，正在加载大量数据。我已经在所有浏览器中进行了测试(甚至尝试过 CURL)。
我在这里缺少什么？

Answer 1

阻止 mapMaterializedValue会这样做，因为它在 Akka default-dispatcher 中运行线程，从而在持续时间内阻止消息路由(详见 this answer)。您希望异步调度缓慢的阻塞代码，并为其发布消息的参与者引用。如果您将来运行它，您的示例将按照您的预期进行:

public CompletionStage<Result> test() {
    return CompletableFuture.supplyAsync( () ->
            Source.<ByteString>actorRef(256, OverflowStrategy.dropNew())
                    .mapMaterializedValue(sourceActor -> {

                        CompletableFuture.runAsync(() -> {

                            for (int i = 0; i < 20; i++) {
                                sourceActor.tell(ByteString.fromString(String.valueOf(i) + "<br/>\n"), null);
                                try {
                                    Thread.sleep(500);//intentional delay
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                }
                            }
                            sourceActor.tell(new Status.Success(NotUsed.getInstance()), null);
                        });

                        return sourceActor;
                    })
    ).thenApplyAsync( chunks -> ok().chunked(chunks).as("text/html"));
}