48.5. 비동기 응답

서버 측에서 비동기 호출을 처리하는 목적은 스레드를 보다 효율적으로 사용할 수 있도록 하고 궁극적으로 서버의 요청 스레드가 모두 차단되기 때문에 클라이언트 연결 시도가 거부되는 시나리오를 방지하는 것입니다. 호출이 비동기적으로 처리되면 요청 스레드가 거의 즉시 해제됩니다.

그림 48.1. “비동기 처리를 위한 스레딩 모델” 서버 측에서 비동기 처리를 위한 기본 모델에 대한 개요를 보여줍니다.

그림 48.1. 비동기 처리를 위한 스레딩 모델

개요에서 요청은 비동기 모델에서 다음과 같이 처리됩니다.In outline, a request is processed as follows in the asynchronous model:

java.util.concurrent API는 매우 쉽게 전체 스레드 풀 구현을 생성할 수 있는 강력한 API입니다. Java 동시성 API의 용어에서 스레드 풀을 execut or라고 합니다. 이는 작업 스레드 및 해당 스레드를 제공하는 차단 대기열을 포함하여 전체 작업 스레드 풀을 만들려면 단일 코드 줄만 필요합니다.

예를 들어 그림 48.1. “비동기 처리를 위한 스레딩 모델” 에 표시된 Executor Thread Pool 과 같은 전체 작업 스레드 풀을 생성하려면 다음과 같이 java.util.concurrent.Executor 인스턴스를 만듭니다.

Executor executor = new ThreadPoolExecutor(
    5,                                    // Core pool size
    5,                                    // Maximum pool size
    0,                                    // Keep-alive time
    TimeUnit.SECONDS,                     // Time unit
    new ArrayBlockingQueue<Runnable>(10)  // Blocking queue
);

이 생성자는 스레드 5개가 포함된 새 스레드 풀을 만들고, 단일 차단 대기열에 의해 제공되며 최대 10개의 Runnable 개체를 보유할 수 있습니다. 스레드 풀에 작업을 제출하려면 executor.execute 메서드를 호출하여 실행 가능한 개체( asynchronous 작업을 캡슐화하는 실행 가능 오브젝트)에 대한 참조를 전달합니다.

비동기적인 리소스 메서드를 정의하려면 @Suspended 주석을 사용하여 javax.ws.rs.container.AsyncResponse 유형의 인수를 주입하고 메서드가 void 를 반환하는지 확인합니다. 예를 들면 다음과 같습니다.

// Java
...
import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.container.AsyncResponse;
import javax.ws.rs.container.Suspended;

@Path("/bookstore")
public class BookContinuationStore {
    ...
    @GET
    @Path("{id}")
    public void handleRequestInPool(@PathParam("id") String id,
                                    @Suspended AsyncResponse response) {
        ...
    }
    ...
}

삽입된 AsyncResponse 오브젝트가 나중에 응답을 반환하는 데 사용되므로 리소스 메서드가 void 를 반환해야 합니다.

javax.ws.rs.container.AsyncResponse 클래스는 들어오는 클라이언트 연결에서 추상 핸들을 제공합니다. AsyncResponse 오브젝트가 리소스 메서드에 삽입되면 기본 TCP 클라이언트 연결이 초기에 일시 중지된 상태입니다. 나중에 응답을 반환할 준비가 되면 기본 TCP 클라이언트 연결을 다시 활성화하고 AsyncResponse 인스턴스에서 resume 를 호출하여 응답을 다시 전달할 수 있습니다. 또는 호출을 중단해야 하는 경우 AsyncResponse 인스턴스에서 취소 를 호출할 수 있습니다.

그림 48.1. “비동기 처리를 위한 스레딩 모델” 에 표시된 비동기 처리 시나리오에서는 일시 중단된 요청을 전용 스레드 풀에서 나중에 처리할 수 있는 큐로 푸시합니다. 그러나 이러한 접근 방식이 작동하려면 개체에서 일시 중단된 요청 을 캡슐화하는 방법이 필요합니다. 일시 중단된 요청 오브젝트는 다음 사항을 캡슐화해야 합니다.

이러한 사항을 캡슐화하는 편리한 방법은 Runnable 클래스를 정의하여 일시 중지된 요청을 나타내는 것입니다. 여기서 Runnable.run() 메서드는 호출 논리를 캡슐화하는 것입니다. 가장 우아한 방법은 다음 예제와 같이 로컬 클래스로 Runnable 을 구현하는 것입니다.

비동기 처리 시나리오를 구현하려면 리소스 메서드 구현에서 실행 가능한 실행 가능 오브젝트를 executor 스레드 풀에 전달해야 합니다.To implement the asynchronous processing scenario, the implementation of the resource method must pass a Runnable object (representing the suspended request) to the executor thread pool. Java 7 및 8에서는 다음 예제와 같이 일부 새로운 구문을 사용하여 Runnable 클래스를 로컬 클래스로 정의할 수 있습니다.

// Java
package org.apache.cxf.systest.jaxrs;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import javax.ws.rs.GET;
import javax.ws.rs.NotFoundException;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;
import javax.ws.rs.container.AsyncResponse;
import javax.ws.rs.container.CompletionCallback;
import javax.ws.rs.container.ConnectionCallback;
import javax.ws.rs.container.Suspended;
import javax.ws.rs.container.TimeoutHandler;

import org.apache.cxf.phase.PhaseInterceptorChain;

@Path("/bookstore")
public class BookContinuationStore {

    private Map<String, String> books = new HashMap<String, String>();
    private Executor executor = new ThreadPoolExecutor(5, 5, 0, TimeUnit.SECONDS,
                                        new ArrayBlockingQueue<Runnable>(10));

    public BookContinuationStore() {
        init();
    }
    ...
    @GET
    @Path("{id}")
    public void handleRequestInPool(final @PathParam("id") String id,
                                    final @Suspended AsyncResponse response) {
        executor.execute(new Runnable() {
            public void run() {
                // Retrieve the book data for 'id'
                // which is presumed to be a very slow, blocking operation
                // ...
                bookdata = ...
                // Re-activate the client connection with 'resume'
                // and send the 'bookdata' object as the response
                response.resume(bookdata);
            }
        });
    }
    ...
}

리소스 메서드 인수, id 및 response 가 Runnable 로컬 클래스 정의로 직접 전달되는 방법을 확인합니다. 이 특수 구문을 사용하면 로컬 클래스에서 해당 필드를 정의하지 않고도 Runnable.run() 메서드에서 직접 리소스 메서드 인수를 사용할 수 있습니다.

비동기 처리 모델은 REST 호출 시 시간 초과도 지원합니다. 기본적으로 시간 초과로 인해 HTTP 오류 응답이 클라이언트로 다시 전송됩니다. 그러나 시간 초과 처리기 콜백을 등록하는 옵션도 있으므로 시간 초과 이벤트에 대한 응답을 사용자 지정할 수 있습니다.But you also have the option of registering a timeout handler callback, which enables you to customize the response to a timeout event.

시간 초과를 지정하지 않고 간단한 호출 타임아웃을 정의하려면 다음 예제와 같이 AsyncResponse 오브젝트에서 setTimeout 메서드를 호출합니다.

// Java
// Java
...
import java.util.concurrent.TimeUnit;
...
import javax.ws.rs.GET;
import javax.ws.rs.NotFoundException;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;
import javax.ws.rs.container.AsyncResponse;
import javax.ws.rs.container.Suspended;
import javax.ws.rs.container.TimeoutHandler;

@Path("/bookstore")
public class BookContinuationStore {
    ...
    @GET
    @Path("/books/defaulttimeout")
    public void getBookDescriptionWithTimeout(@Suspended AsyncResponse async) {
        async.setTimeout(2000, TimeUnit.MILLISECONDS);
        // Optionally, send request to executor queue for processing
        // ...
    }
    ...
}

java.util.concurrent.TimeUnit 클래스에서 언제든지 시간 단위를 사용하여 시간 초과 값을 지정할 수 있습니다. 위 예제는 요청을 executor 스레드 풀로 보내는 코드를 표시하지 않습니다. 시간 제한 동작을 테스트하려는 경우 리소스 메서드 본문에 async.SetTimeout 호출만 포함하고 모든 호출에 시간 초과가 트리거될 수 있습니다.

AsyncResponse.NO_TIMEOUT 값은 무한한 타임아웃을 나타냅니다.

기본적으로 호출 타임아웃이 트리거되면 JAX-RS 런타임에서 ServiceUnavailableException 예외를 발생시키고 상태 503 을 사용하여 HTTP 오류 응답을 반환합니다.

시간 제한 동작을 사용자 정의하려면 TimeoutHandler 인터페이스를 구현하여 시간 초과 처리기를 정의해야 합니다.

// Java
package javax.ws.rs.container;

public interface TimeoutHandler {
    public void handleTimeout(AsyncResponse asyncResponse);
}

구현 클래스에서 handleTimeout 메서드를 재정의하면 시간 제한을 처리하기 위해 다음 방법 중 하나를 선택할 수 있습니다.

시간제한 처리기를 사용하여 호출 시간을 정의하려면 다음 예제와 같이 AsyncResponse 개체에서 setTimeout 메서드 및 setTimeoutHandler 메서드를 모두 호출합니다.

// Java
...
import javax.ws.rs.GET;
import javax.ws.rs.NotFoundException;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;
import javax.ws.rs.container.AsyncResponse;
import javax.ws.rs.container.Suspended;
import javax.ws.rs.container.TimeoutHandler;

@Path("/bookstore")
public class BookContinuationStore {
    ...
    @GET
    @Path("/books/cancel")
    public void getBookDescriptionWithCancel(@PathParam("id") String id,
                                             @Suspended AsyncResponse async) {
        async.setTimeout(2000, TimeUnit.MILLISECONDS);
        async.setTimeoutHandler(new CancelTimeoutHandlerImpl());
        // Optionally, send request to executor queue for processing
        // ...
    }
    ...
}

여기서 이 예제에서는 호출 타임아웃을 처리하기 위해 CancelTimeoutHandlerImpl 타임아웃 핸들러의 인스턴스를 등록합니다.

CancelTimeoutHandlerImpl 시간제한 핸들러는 다음과 같이 정의됩니다.

// Java
...
import javax.ws.rs.container.AsyncResponse;
...
import javax.ws.rs.container.TimeoutHandler;

@Path("/bookstore")
public class BookContinuationStore {
    ...
    private class CancelTimeoutHandlerImpl implements TimeoutHandler {

        @Override
        public void handleTimeout(AsyncResponse asyncResponse) {
            asyncResponse.cancel();
        }

    }
    ...
}

AsyncResponse 오브젝트에 대한 호출 취소 효과는 HTTP 503 (서비스 사용할 수 없음) 오류 응답을 클라이언트에 보내는 것입니다. 응답 메시지에 Retry-After: HTTP 헤더를 설정하는 데 사용되는 cancel 메서드( int 또는 java.util.Date 값)에 대한 인수를 선택적으로 지정할 수 있습니다. 그러나 클라이언트는 종종 Retry-After: 헤더를 무시합니다.

실행 가능 스레드 풀에서 처리를 위해 대기 중인 실행 가능한 인스턴스로 캡슐화된 경우 스레드 풀이 요청을 처리하는 시점에 AsyncResponse가 취소되었음을 확인할 수 있습니다.If you have encapsulated a suspended request as a Runnable instance, which is queued for processing in an executor thread pool, you might find that the AsyncResponse has been canceled by the time the thread pool gets around to processing the request. 이러한 이유로 Runnable 인스턴스에 일부 코드를 추가하여 취소된 AsyncResponse 개체에 대처할 수 있습니다. 예를 들면 다음과 같습니다.

// Java
...
@Path("/bookstore")
public class BookContinuationStore {
    ...
    private void sendRequestToThreadPool(final String id, final AsyncResponse response) {

        executor.execute(new Runnable() {
            public void run() {
                if ( !response.isCancelled() ) {
                    // Process the suspended request ...
                    // ...
                }
            }
        });

    }
    ...
}

클라이언트 연결이 끊어진 경우를 처리하기 위해 콜백을 추가할 수 있습니다.

삭제된 연결에 대한 콜백을 추가하려면 다음과 같이 정의된 javax.ws.rs.container.ConnectionCallback 인터페이스를 구현해야 합니다.

// Java
package javax.ws.rs.container;

public interface ConnectionCallback {
    public void onDisconnect(AsyncResponse disconnected);
}

연결 콜백을 구현한 후 레지스터 방법 중 하나를 호출하여 현재 AsyncResponse 오브젝트에 등록해야 합니다. 예를 들어, type의 연결 콜백을 등록하려면 MyConnectionCallback:

asyncResponse.register(new MyConnectionCallback());

일반적으로 연결 콜백을 구현하는 주요 이유는 삭제된 클라이언트 연결과 관련된 리소스를 확보할 수 있습니다(여기서 AsyncResponse 인스턴스를 사용해야 하는 리소스를 식별하는 키로 사용할 수 있음).

호출이 완료되면 알림을 받기 위해 선택적으로 AsyncResponse 인스턴스에 콜백을 추가할 수 있습니다. 이 콜백을 호출할 수 있는 경우 처리에는 다음 두 가지 대체 지점이 있습니다.

완료 콜백을 추가하려면 다음과 같이 정의된 javax.ws.rs.container.CompletionCallback 인터페이스를 구현해야 합니다.

// Java
package javax.ws.rs.container;

public interface CompletionCallback {
    public void onComplete(Throwable throwable);
}

일반적으로 throw 가능한 인수는 null 입니다. 그러나 요청 처리가 매핑되지 않은 예외가 발생한 경우 throw할 수 없는 예외 인스턴스가 throw됩니다.However, if the request processing resulted in an unmapped exception, throwable contains the unmapped exception instance.

완료 콜백을 구현한 후 레지스터 방법 중 하나를 호출하여 현재 AsyncResponse 오브젝트에 등록해야 합니다. 예를 들어, type 완료 콜백을 등록하려면 MyCompletionCallback:

asyncResponse.register(new MyCompletionCallback());