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의 용어에서 스레드 풀은 executor 라고 합니다. 작업 스레드와 피드를 제공하는 차단 대기열을 포함하여 전체 작업 스레드 풀을 생성하려면 한 줄의 코드만 필요합니다.

예를 들어 그림 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
);

이 생성자는 최대 10개의 실행 가능한 개체를 보유할 수 있는 단일 차단 큐로 5개의 스레드가 있는 새 스레드 풀을 생성합니다. 작업을 스레드 풀에 제출하려면 executor.execute 메서드를 호출하여 비동기 작업을 캡슐화하는 실행 가능한 개체에 대한 참조를 전달합니다.

비동기적인 리소스 메서드를 정의하려면 @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 클라이언트 연결이 처음에 일시 중지된 상태입니다. 나중에 응답을 반환할 준비가 되면 AsyncResponse 인스턴스에서 resume 를 호출하여 기본 TCP 클라이언트 연결을 다시 활성화하고 응답을 다시 전달할 수 있습니다. 또는 호출을 중단해야 하는 경우 AsyncResponse 인스턴스에서 cancel 을 호출할 수 있습니다.

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

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

비동기 처리 시나리오를 구현하려면 리소스 메서드의 구현에서 실행 가능한 개체(부팅된 요청을 표시)를 executor 스레드 풀에 전달해야 합니다. 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 오류 응답이 클라이언트에 다시 전송됩니다. 그러나 시간 초과 이벤트에 대한 응답을 사용자 지정할 수 있는 시간 초과 처리기 콜백을 등록할 수도 있습니다.

간단한 호출 시간 초과를 정의하려면 다음 예와 같이 시간 초과 처리기를 지정하지 않고 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 값은 무한 타임아웃을 나타냅니다.

기본적으로 호출 제한 시간이 트리거되면 Cryostat-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 오브젝트에서 cancel 을 호출하면 HTTP 503(서비스를 사용할 수 없음) 오류 응답을 클라이언트에 전송하는 것입니다. 선택적으로 응답 메시지에서 Retry-After: HTTP 헤더를 설정하는 데 사용되는 cancel 메서드( int 또는 java.util.Date 값)에 인수를 지정할 수 있습니다. 그러나 클라이언트는 종종 Retry-After: 헤더를 무시합니다.

executor 스레드 풀에서 처리를 위해 대기 중인 요청을 실행 가능한 인스턴스로 캡슐화한 경우 스레드 풀이 요청을 처리하는 동안 AsyncResponse 가 취소되었음을 확인할 수 있습니다. 이러한 이유로 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);
}

연결 콜백을 구현한 후 register 방법 중 하나를 호출하여 현재 AsyncResponse 오브젝트에 등록해야 합니다. 예를 들어 다음과 같은 유형의 연결 콜백을 등록하려면 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);
}

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

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

asyncResponse.register(new MyCompletionCallback());