Vert.x 프레임워크(3.X) 이벤트 핸들러 처리(1), 2019-12-17 Vert.x Walker

Vert.x는 Java 진영에 고성능 서버 프레임워크 중 하나이다. 그리고 Vert.x 프레임워크는 Event Driven 프로그래밍 모델을 사용하고 non blocking인 것을 강조한다. 이는 개발자가 특정 이벤트가 발생했을때 호출될 이벤트 핸들러를 수 없의 정의하고 이것들이 블럭킹 없이 비동기로 동작한다는 예상할 수 있다. 이렇게 Vert.x는 Event Driven/Non blocking 아키텍처 위에서 고성능 비동기 서버 프로그래밍을 할 수 있다.
Vert.x는 비동기 프로그래밍을 Call-Back 패턴으로 구현한다. Vert.x에서 이 Call-Back 패턴을 어떤 식으로 구현해서 사용하는지 알아보자.(Vert.x는 Reactive Programming도 지원함)

이글은 내용이 길어져서 두개의 파트로 나누어 작성하였다.

글 목록.

Vert.x 프레임워크 이벤트 핸들러 처리. 1
Vert.x 프레임워크 이벤트 핸들러 처리. 2

• 예제 코드에 비동기 동작을 위한 쓰레드 구현 부분은 제외하였다.

Handler<T>

콜백 함수라고 하면 개발자라면 한 번씩은 들어 봤을 것이다. 말 그대로 어떤 작업이 완료되었을 때 특정 함수를 호출하도록 지정하는 것이다.
Vert.x에서는 이런 콜백을 핸들러라고 명칭하고 별도의 Interface를 제공한다. Handler<T> 인터페이스이다.

/**
 *  A generic event handler.
 *  <p>
 *  This interface is used heavily throughout Vert.x as a handler for all types of asynchronous occurrences.
 *  <p>
 *
 *  @author <a href="http://tfox.org">Tim Fox</a>
 */
@FunctionalInterface
public interface Handler<E> {

  /**
   * Something has happened, so handle it.
   *
   * @param event  the event to handle
   */
  void handle(E event);
}

Interface를 정의 내용을 보면 단순히 [제너릭 타입 인자를 하나 받은 메소드]를 가진 함수형 인터페이스이다.
이 핸들러 인터페이스는 Vert.x 코드 전체에서 람다 함수와 같이 형태로 아주 유용하게 사용된다.

다음 타이머 이벤트 메소드에서 핸들러를 사용하는 예를 보자.

/**
 * Set a periodic timer to fire every {@code delay} milliseconds, at which point {@code handler} will be called with
 * the id of the timer.
 *
 *
 * @param delay  the delay in milliseconds, after which the timer will fire
 * @param handler  the handler that will be called with the timer ID when the timer fires
 * @return the unique ID of the timer
 */
long setPeriodic(long delay, Handler<Long> handler);

setPeriodic 메소드의 시그니처를 보면 알 수 있듯이 타이머 딜레이 타임과 Handler<Long> 타입의 클래스를 인자로 받는 걸로 정의되어 있다.
실제 사용 예를 보자.

vertx.setPeriodic(1000 * 5 , id -> Sytem.out.println("Fire!"));

위 코드는 5초 간격으로 id -> Sytem.out.println("Fire!") 람다 함수를 호출하게 된다. 이때 id는 호출 유니크 아이디 값이라 해당 코드에서는 의미가 없다.

Vert.x 프레임워크에서 비동기 동작을 하는 거의 모든 메소드는 위와 같은 기본을 형태 취한다.
메소드가 동작하기 위한 인자를 받고, 비동기로 동작하고, 마지막으로 해당 메소드의 비동기 동작의 결과(T)를 Handler 제너릭 타입 오브젝트로 받아 처리한다.
위 구조를 말로 쉽게 풀어쓰면 “메소드 로직을 비동기로 실행하고, 로직이 완료되었을 경우 완료 값(T Type)을 Handler.handler() 메소드의 인자 값으로 넘기면서 실행해줘!” 이 정도 일 것이다.
결과적으로 나중에 실행할 메소드를 전달하는 것이다. 결과값의 형태를 한정 할 수 없기에 실행 메소드 시그니처를 제러릭타입 T로 지정하는 것이다.
이런 Handler 인터페이스 때문에 콜백 함수를 Vert.x에서는 쉽고 직관적으로 구현할 수 있다.

AsyncResult<T>

위에서 살펴본 Handler 인터페이스는 기본적으로 콜백 기능을 구현하는데 많은 도움이 된다. 하지만 비동기 로직 상에서 알 수 없는 문제가 발생했을 경우는 어떻게 처리해야 할까? 처리과정에 예외가 발생하여 결과 값이 없으면 어떻게 될까? 단순하게 Handler 인터페이스에 null을 넘겨주는 것은 가능할 것이다. 하지만 개발자 입장에서 왜 예외나 문제가 발생했는지 정보를 확인하고 싶을 것이다. 그래서 Vert.x에서는 AsyncResult<T>라는 비동기 처리 결과값을 전달하기 위한 인테페이스를 제공한다.
Handler<T>가 비동기 처리 완료시 동작을 정의하는 것이라면 AsyncResult<T>는 비동기 처리 상태와 결과값을 저장 장소 것이다. 즉 Handler의 제너릭타입 T를 한번 AsyncResult객체로 감싸서 처리하는 것이다. 실제 사용 시 Handler<AsyncResult<T>>형태로 사용하게 된다.

public interface AsyncResult<T> {
  T result();
  Throwable cause();
  boolean succeeded();
}

위 소스의 내용은 AsyncResult<T> 인터페이스에서 자주 사용되는 메소드이다.
세부 기능을 보면 아래와 같다.

• succeeded() : 로직이 성공했는지 체크하기 위해 사용.
• result() : 성공했을 경우에 로직의 결과 값을 반환.
• cause() : 예외가 발생할 경우 해당 예외 객체 정보.

필요한 경우에 구현체를 정의해서 반환하면 된다.

그럼 실 사용 예제를 보자.

private void callBackTestMethod(Handler<AsyncResult<String>> handler) {

    String resultString = "this is test.";
    Throwable th = null;
    AsyncResult<String> asyncResult = new AsyncResult<String>() {
        public String value = resultString;
        public void complete(String value) {
            this.value = value;
        }

        @Override
        public String result() {
            return value;
        }

        @Override
        public Throwable cause() {
            return th;
        }

        @Override
        public boolean succeeded() {
            return value != null ? true : false;
        }

        public boolean failed() {
            return value != null ? false : true;
        }
    };
   // prcess end.
   handler.handle(asyncResult);
}

코드 자체가 실 사용하기에는 문제가 있지만 흐림만 보기로 하자.

• 코드에서 보면 핸들러 메쏘드의 매개변수로 Handler<AsyncResult<String>> handler 받는다.

  풀어쓰면 [AsyncResult 매개변수]를 받는 함수형 인터페이스 객체를 매개변수로 받는다.

• 그리고 실제로 메소드 내부에서 만들어진 AsyncResult<String>의 구현체를 핸들러로 넘겨주는 것을 볼 수 있다.

실제 프레임워크에서 구현된 이와 같은 형태의 메소드를 호출하는 코드는 아래와 같은 형태로 나올 수 있다.

callBackTestMethod(ar->{
    if (ar.succeeded()) {
        String result = ar.result();
        System.out.println("result:" + result);
    } else {
        System.err.println("error:"+ar.cause().getMessage());
    }
});

코드에서도 알 수 있듯이 실제로 호출하는 곳에서 Handler을 생성하고, Handler에 내부에서 인자로 넘어오는 AsyncResult 객체에서 성공/실패 여부와 성공 시 결과 값도 가져올 수 있다.

Future<T>

위에서 알아본 Handler<T>와 AsyncResult<T>는 Vert.x 프레임워크 소스에 많은 부분을 차지한다. 하지만 실제 사용하기 위해서는 반복적인 코드가 많이 필요하다. 인터페이스이기 때문에 별도의 구현체를 다 정의해야 한다. 예를 들어 단순하게 성공/실패만 확인하기 위해서도 모든 내용을 구현해야 한다. 그래서 이런 번거로움을 해결하기 위해서 Vert.x는 Future 클래스를 제공한다.
Future 중요한 몇 개의 메소드만 보자.

public interface Future<T> extends AsyncResult<T>, Handler<AsyncResult<T>> {
  /**
   * Create a future that hasn't completed yet
   *
   * @param <T>  the result type
   * @return  the future
   * @deprecated instead use {@link Promise#promise()}
   */
  @Deprecated
  static <T> Future<T> future() {
    return factory.future();
  }
  
    /**
   * Created a succeeded future with the specified result.
   *
   * @param result  the result
   * @param <T>  the result type
   * @return  the future
   */
  static <T> Future<T> succeededFuture(T result) {
    if (result == null) {
      return factory.succeededFuture();
    } else {
      return factory.succeededFuture(result);
    }
  }
  
    /**
   * Like {@link #onComplete(Handler)}.
   */
  @Fluent
  Future<T> setHandler(Handler<AsyncResult<T>> handler);
  
    /**
   * Set the result. Any handler will be called, if there is one, and the future will be marked as completed.
   *
   * @param result  the result
   * @deprecated instead create a {@link Promise} and use {@link Promise#complete(Object)}
   */
  @Deprecated
  void complete(T result);
  
    /**
   * Set the failure. Any handler will be called, if there is one, and the future will be marked as completed.
   *
   * @param cause  the failure cause
   * @deprecated instead create a {@link Promise} and use {@link Promise#fail(Throwable)}
   */
  @Deprecated
  void fail(Throwable cause);
}

우선 Future 클래스는 AsyncResult<T>, Handler<AsyncResult<T>>를 상속하고 있다. 비동기 결과 및 상태이면서 비동기 결과를 처리하는 콜백 메소드 클래스이기도 한 것이다. 이 말은 핸들러를 만들면서 비동기 결과까지 한 번에 더 정의할 수 있다는 것이다. Future<T>객체 전달해줄테니 비동기 결과값도 저장하고 비동기 작업이 끝나면 이 객체의 핸들러 메소드를 호출해달라는 것이다.
그리고 위에 메소드 중에 future() / succeededFuture(T result) 같은 팩토리 메소드를 가지고 있어서 구현체를 바로 생성하거나 성공한 AsyncResult<T> 를 가지는 Future 클래스를 만들 수 도 있다.
한발 더 나아가 이미 만들어진 Future 클래스에 AsyncResult<T>을 변경할 수 있는 complete(T result) / fail(Throwable cause)를 제공한다.
마지막으로 setHandler(Handler<AsyncResult<T>> handler) 메소드로 핸들러 클래스 구현체를 외부에서 지정할 수도 있다.
결과적으로 단 몇 줄로 핸들러와 비동기 반환 값 객체를 만들어 낼 수 있다.
간단한 예제 코드를 보자.

private void testProcess(Future<String> stringFuture) {
    ...
    try {
        stringFuture.complete("this is test.");
    } catch (Exception ex) {
        stringFuture.fail(ex);
    }
}

private void caller() {
	 // 문자를 결과로 받는 객체 생성.
    Future<String> stringFuture = Future.future();
    
    // Future 객체를 받아서 내부 처리 결과를 저장함.
    testProcess(stringFuture);
    
    // testHander의 실행 결과를 처리할 수 있음.
    stringFuture.setHandler(ar->{
        if (ar.succeeded()) {
            String result = ar.result();
            System.out.println("result:" + result);
        } else {
            System.err.println("error:"+ar.cause().getMessage());
        }
    });
}

위 코드에서 보면 인터페이스 구현 부분이 전부 사라졌다. 그리고 코드 자체가 더 직관적으로 보임을 알 수 있다.

Promise<T>

Future 인터페이스는 Vert.x 프레임워크에서 없어서는 안 될 중요한 클래스이다. 그런 Future 인터페이스에 최신 버전에 대대적인 수정이 가해졌다.
Future 인터페이스에서 내부 값을 변경하는 메소드가 Deprecated 되고 대신 Promise 인터페이스가 추가되었다.

Promise는 중요 메소드는 아래와 같다.

public interface Promise<T> extends Handler<AsyncResult<T>> {
  static <T> Promise<T> promise() {
    return factory.promise();
  
  }
  void complete(T result);
  void fail(Throwable cause);
  Future<T> future();
  ...
}

기본 Handler<AsyncResult<T>>를 상속하고 있고, Future<T> future() 팩토리 메소드를 가지고 있다.
complete/fail 같은 Furture 인터페이스에서 Deprecated된 메소드로 구성되어 있다.
사용법은 Promise 인터페이스 객체를 생성해서 값을 쓰고 Future로 변환해서 값을 조회하는 방법으로 사용된다.
C++ Future/Promise 클래스와 유사하게 변경된 것 같다. C++ 비동기 처리 시에 Promise로 Future객체를 유도하고 비동기 쓰레드에서 Promise 객체로 값을 저장하고 비동기 쓰레드가 완료되면 Future로 값을 조회한다.
기능상으로 Promise는 쓰기 담당/Future은 읽기 담당으로 변경된 것이다. 사견으로는 기존 Future로 쓰고 읽고 다 할 경우 비동기 구간에서 생성 된 데이터가 완료 전에 변경될 수 있는 문제(비동기 프로세스의 동시 데이터 접근 문제)를 방지하기 위해서 나눈 것 같기도 하다.

사용 예제 코드를 보자.

private Future<String> testPorcess() {
    Promise<String> stringPromise = Promise.promise();
    try {
        stringPromise.complete("this is test.");
    } catch (Exception ex) {
        stringPromise.fail(ex);
    }
    return stringPromise.future();
}

private void caller() {
    testPorcess().setHandler(ar->{
        if (ar.succeeded()) {
            String result = ar.result();
            System.out.println("result:" + result);
        } else {
            System.err.println("error:"+ar.cause().getMessage());
        }
    });
}

기존 Future와 달라진 코드는 Promise를 값을 저장하고 .future() 메소드를 호출해서 Future로 반환해서 처리하는 부분만 다르고 동일하다.

마무리

Vert.x 프레임워크는 콜백으로 비동기를 처리하기 때문에 ‘콜백 지옥’ 빠지기 쉽다. 하지만 Handler, AsyncResult, Future, Promise 인터페이스를 잘 활용하면 비동기 처리 콜백 패턴의 문제점을 해결 할 수 있다. 마치 Reactive 프로그래밍의 Subscribe와 유사하게 구현할 수 있다.
해당 내용은 다음 글에서 설명하겠다.