Понимание Spring Reactive: Servlet Async

В предыдущей статье мы обсуждали, как развивались контейнеры сервлетов, и превратили связь между клиентом и сервером в неблокирующую парадигму. В этой статье мы сосредоточимся на эволюции Java-сервлетов (и Spring) в направлении неблокирующего, реактивного мира.

Давайте вспомним запрос потока при получении разъемом NIO:

1. Несколько потоков (1-4 в зависимости от количества ядер) опрашивают селектор, ища активность ввода-вывода на канале соединения.
2. Когда селектор видит активность ввода-вывода, он вызывает  handle метод для соединения, и потоку  pool отводится для обработки.
3.  Thread попытается прочитать соединение и проанализировать его и для HTTP-соединения. Если заголовки запроса завершены, поток продолжает вызывать обработку запроса (в конце концов, он попадает в сервлет) без ожидания какого-либо содержимого.
4. Как только a  thread отправляется сервлету, он выглядит так, как будто IO сервлета блокирует, и, следовательно, любая попытка чтения / записи данных из   HttpInputStream /  HttpOutputStream должен блокироваться. Но, поскольку мы используем нижний соединитель NIO, операции ввода-вывода используются  HttpInputStream и  HttpOutputStream являются асинхронными с обратными вызовами. Из-за блокирующего характера Servlet API он использует специальный обратный вызов блокировки для достижения блокировки.

Шаг 4, приведенный выше, разъяснил бы больше термин «имитация блокировки», использованный в предыдущей статье.

Проблемы перед Servlet 3.0

Теперь, возвращаясь к проблемам, которые создает один поток на модель запроса, мы видим, что фактическая обработка запроса, которая по своей природе является блокирующей, выполняется потоком (назовем его потоком запроса) из пула, которым управляет контейнер сервлет. В NIO размер пула потоков по умолчанию равен 200, что означает, что одновременно может обслуживаться только 200 запросов. Проблема с синхронной обработкой запросов заключается в том, что потоки (выполняющие тяжелую работу) выполняются в течение длительного времени, прежде чем отклик исчезнет. Если это происходит в масштабе, в контейнере сервлета со временем заканчиваются потоки — длительные потоки приводят к истощению потоков.

Этот размер может быть увеличен для соответствия любому числу (с аппаратными ограничениями), но тогда это также приведет к накладным расходам на переключение контекста, очистку кеша и т. Д. Хотя увеличение потоков и обслуживание большего количества параллельных запросов не является плохой идеей, в случае, если приложение требует высокого параллелизма, тогда нам нужно найти какой-то другой подходящий подход. Давайте продолжим читать, чтобы лучше понять подход обработки большего количества одновременных пользователей без увеличения размера пула потоков контейнера.

Серверный поток заблокирован во время обработки Http-запроса.

Асинхронные Сервлеты в 3.0

Асинхронный сервлет позволяет приложению обрабатывать входящие запросы в асинхронном режиме. Данный поток HTTP-запроса обрабатывает входящий запрос, а затем передает запрос другому фоновому потоку, который, в свою очередь, будет отвечать за обработку запроса и отправляет ответ обратно клиенту. Исходный поток HTTP-запроса вернется в пул HTTP-потока, как только он передаст запрос в фоновый поток, поэтому он станет доступным для обработки другого запроса.

Серверный поток освобождается во время обработки запроса Http

Ниже приведен фрагмент кода о том, как этого можно достичь в Servlet 3.0:

@WebServlet(name="myServlet", urlPatterns={"/asyncprocess"}, asyncSupported=true)
public class MyServlet extends HttpServlet {
    public void doGet(HttpServletRequest request, HttpServletResponse response) {
        OutputStream out = response.getOutputStream();
        AsyncContext aCtx = request.startAsync(request, response);
        //process your request in a different thread
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                String json ="json string";
                out.write(json);
                ctx.complete();
            }
        };
        //use some thread pool executor
        poolExecutor.submit(runnable);
    }
}

 @GetMapping(value = "/blockingRequestProcessing")

    public String blockingRequestProcessing() {

        logger.debug("Blocking Request processing Triggered");

        String url = "http://localhost:8090/sleep/1000";

        new RestTemplate().getForObject(url, Boolean.TYPE);

        return "blocking...";

    }

    @GetMapping(value = "/asyncBlockingRequestProcessing")

    public CompletableFuture<String> asyncBlockingRequestProcessing(){

        return CompletableFuture.supplyAsync(() -> {

            logger.debug("Async Blocking Request processing Triggered");

            String url = "http://localhost:8090/sleep/1000";

            new RestTemplate().getForObject(url, Boolean.TYPE);

            return "Async blocking...";

        },asyncTaskExecutor);

    }

@GetMapping(value = "/asyncNonBlockingRequestProcessing")
    public CompletableFuture<String> asyncNonBlockingRequestProcessing(){
            ListenableFuture<String> listenableFuture = getRequest.execute(new AsyncCompletionHandler<String>() {
                @Override
                public String onCompleted(Response response) throws Exception {
                    logger.debug("Async Non Blocking Request processing completed");
                    return "Async Non blocking...";
                }
            });
            return listenableFuture.toCompletableFuture();
    }