爲什麼玩2.5 Akka大塊響應一次加載全部

Question

我試圖在Akka中使用PLay 2在webapp中實現大塊響應。但是，不是按塊加載響應，而是所有響應都是一次。下面是由我在控制器中創建塊代碼：

/** 
* 
*/  
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 { 
    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" 
} 

正如你可以看到之前發送最後一個大塊我有意拖延響應時間。因此在邏輯上，所有分塊的數據都應在它之前交付。
但是，在我的情況下，大量的數據正在被加載。我已經在所有瀏覽器中測試過（甚至嘗試過CURL）。
我在這裏錯過了什麼？

Answer 1

在mapMaterializedValue阻塞會做，因爲它在阿卡default-dispatcher線程運行，從而防止消息路由的時間（見this answer瞭解詳細信息）。您想要異步調度緩慢的阻塞代碼，並使用actor引用來發送消息。你的榜樣會做，如果你在以後的運行它，你想到了什麼：

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")); 
} 

Answer 2

如果檢查Source代碼，你可以看到，第一個參數是緩衝區大小

public static <T> Source<T,ActorRef> actorRef(int bufferSize, 
               OverflowStrategy overflowStrategy) 

所有的元素，你在流中生成的可能少於256個字節，因此只生成一個http塊。嘗試添加更多元素，如@Mikesname示例。

Answer 3

如果您需要使用其他方法分塊響應，這可能會有用。

public Result test() { 
    try { 
     // Finite list 
     List<String> sourceList = Arrays.asList("kiki", "foo", "bar"); 
     Source<String, ?> source = Source.from(sourceList); 

    /* Following DB call, which fetch a record at a time, and send it as chunked response. 
     final Iterator<String> sourceIterator = Person.fetchAll(); 
     Source<String, ?> source = Source.from(() -> sourceIterator); */ 

     return ok().chunked(source.via(Flow.of(String.class).map(ByteString::fromString))).as(Http.MimeTypes.TEXT); 

    } catch (Exception e) { 
     return badRequest(e.getMessage()); 
    } 
}