제대로, 제작 한 Akka가 <a href="https://www.playframework.com/documentation/2.5.x/JavaStream#Chunked-responses" rel="nofollow noreferrer">Play! Framework documentation</a>에서 제공하는 예제를 사용하여

Question

스트림 일부 지연 이벤트를 발행하는 수정 된 버전을 사용하여 응답을 청크 얻을 수 없다, 그래서 나는 그들에게 클라이언트 측에서 같은 비율로 발생하는 관찰 할 수있다 :

public Result playExampleDelayed() { 
    Source<ByteString, ?> source = Source.<ByteString> actorRef(5, OverflowStrategy.dropNew()) 
      .mapMaterializedValue(sourceActor -> { 
       for (int i = 0; i < 10; ++i) { 
        Thread.sleep(1000); 
        sourceActor.tell(ByteString.fromString("tick " + i), null); 
       } 
       sourceActor.tell(new Status.Success(NotUsed.getInstance()), null); 
       return null; 
      }); 
    return ok().chunked(source); 
} 

그러나 소스가 완료되면 curl을 사용하여 모든 이벤트를 단일 단계에서 얻습니다. 내가 의도 된 동작을 얻을 수있는 다른 소스 유형을 사용

:이 경우

public Result tick() { 
    Source<ByteString, ?> source = Source.<ByteString> tick(Duration.create(0, TimeUnit.SECONDS), 
      Duration.create(1, TimeUnit.SECONDS), ByteString.fromString("tick")); 
    return ok().chunked(source); 
} 

, 나는 초당 콘솔 하나의 덩어리에서 얻을 수 있습니다.

Akka 설명서에 따르면 첫 번째 예제가 작동 할 것으로 기대합니다. 내가 도대체 ​​뭘 잘못하고있는 겁니까?

Answer 1

mapMaterializedValue 전화에서 Thread.sleep을 사용 중입니다. 이것은 스트림 run() 직후에 발생하는 동기식 호출입니다. 여기서 차단 (예 : Thread.sleep)하면 전체 구체화가 차단됩니다. 따라서 모든 메시지는 루프 실행 마지막 단계에서 액터에 의해 선택됩니다.

결론 : Akka를 사용할 때는 항상 Thread.sleep을 사용하지 마십시오.

반대로 Source.tick은 스케줄러 (비동기, 비 차단)를 사용하므로 성능이 뛰어나고 견고하고 세련된 솔루션입니다.