Page History

퓨쳐는 여러함수의 결과가 다음 함수의 입력이되어야할시, 여러 함수가 병렬로 연결되며

나중에 여러 함수의 결과를 조합해야하는 파이프라이닝(pipelining)에 좋은 도구입니다.

var request = EventRequest(ticketNr)    //요청을만든다
val futureRoute : Future[Route] = Future{
  callEventService(request).event
}.map{ event =>
  val trafficRequest = TrafficRequest{
    destination = event.location,
    arrivalTime = event.time
  )
  callTrafficeService(trafficRequest).route  //교통 루트를 반환한다.
}

import java.util.concurrent.CompletableFuture;
 
 
//선언부
@FunctionalInterface
interface FuncB {
    public int calc(int a, int b);
}
  
@FunctionalInterface
interface FuncA {
    public int calc(int a);
}
 
//사용부
FuncA F1 = (int a) -> a+1;
FuncA F2 = (int a) -> a+1;
FuncA F3 = (int a) -> a+1;
FuncB F4 = (int a,int b) -> a+b;
              
Integer input=1;
CompletableFuture<Integer> futureB
        = CompletableFuture.supplyAsync(() -> F1.calc(1) );
 
CompletableFuture<Integer> futureD
        = CompletableFuture.supplyAsync(() -> F3.calc(F2.calc(1))  );
 
CompletableFuture<Void> combinedFuture
        = CompletableFuture.allOf(futureB, futureD)
        .thenAccept( r-> {
            // 최종 연산로직을 비동기로 처리한 케이스
            Integer value1 = futureB.join();
            Integer value2 = futureD.join();
            log( String.format("async result:%d", value1+value2 ));
        });
combinedFuture.join(); //연산을 비동기적으로 시작하려면 join명령을 사용합니다.
// 최종 연산로직을 동기처리로 변환한 예
combinedFuture.get();   //동기처리로 진행하려면 get을 사용합니다.
Integer sresult = futureB.get() + futureD.get();
log( String.format("sync result:%d",sresult) );

def sendTokafka(recode: ProducerRecod): Futrure[RecordMetadata] = {
  val promise : Promise[[RecordMetadata]] = Promise[RocordMetadata]() //RecordMetadata 타잎의 값을 돌려주는 약속을 만든다.
  val future:Future[RecordMetadata] = promise.future  //future에 대한 참조를 가져온다
  
  val callback = new CallBack(){  //송신성공시 카프카 콜백
    def onCompletition(metadata: RecordMetadata, e:Exception): Unit={
      if(e != null ) promise.failure(e)  //오류기록
      else promise.success(metadata)     //성공기록
    }
  }
  producer.send(record,callback)
  future
}

public class ReActor : ReceiveActor
{
  private ILoggingAdapter log = Context.GetLogger();

  public ReActor()
  {
      string myPath = Self.Path.ToString();

      Receive<string>(message => {
          Handle(message);                
      });

      Receive<DelayReply>(message => {
          Handle(message);
      });
  }

  public void Handle(string str)      //InMessage
  {
      Task.Run(async () =>
      {
          await Task.Delay(1000); //동기적함수를 호출하여 지연시키지만 액터는 멈추지 않습니다.
          DelayReply reply = new DelayReply();
          reply.message = str;
          return reply;
      }).PipeTo(Self);
  }

  public void Handle(DelayReply data) //Out
  {
      string logtrace = string.Format("I'am {0} RE:{1}", Self.Path, data.message);
      log.Info(data.message);
      Sender.Tell(data);
  }
          
}