Асинхронное программирование HTTP с помощью платформы Play

Автор оригинала: Gian Mario Contessa.

1. Обзор

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

В этом руководстве мы запустим несколько асинхронных запросов к службе из приложения Play Framework. Используя неблокирующие возможности HTTP Java, мы сможем плавно запрашивать внешние ресурсы, не влияя на нашу собственную основную логику.

В нашем примере мы рассмотрим библиотеку Play WebService .

2. Библиотека Play WebService (WS)

WS-это мощная библиотека, обеспечивающая асинхронные HTTP-вызовы с использованием Java Действие .

Используя эту библиотеку, наш код отправляет эти запросы и продолжает работу без блокировки. Для обработки результата запроса мы предоставляем функцию потребления, то есть реализацию интерфейса Consumer .

Этот шаблон имеет некоторое сходство с реализацией JavaScript обратных вызовов, Promises, и шаблоном async/await .

Давайте построим простой Потребитель , который регистрирует некоторые данные ответа:

ws.url(url)
  .thenAccept(r -> 
    log.debug("Thread#" + Thread.currentThread().getId() 
      + " Request complete: Response code = " + r.getStatus() 
      + " | Response: " + r.getBody() 
      + " | Current Time:" + System.currentTimeMillis()))

Наш Потребитель просто регистрируется в этом примере. Однако потребитель может делать все, что нам нужно, с результатом, например, хранить результат в базе данных.

Если мы заглянем глубже в реализацию библиотеки, то увидим , что WS обертывает и настраивает AsyncHttpClient Java, который является частью стандартного JDK и не зависит от воспроизведения.

3. Подготовьте Пример проекта

Чтобы поэкспериментировать с фреймворком, давайте создадим несколько модульных тестов для запуска запросов. Мы создадим каркасное веб-приложение, чтобы ответить на них, и будем использовать WS framework для выполнения HTTP-запросов.

3.1. Скелетное Веб – Приложение

Прежде всего, мы создаем исходный проект с помощью команды sbt new :

sbt new playframework/play-java-seed.g8

В новой папке мы затем отредактируем файл build.sbt и добавим зависимость библиотеки WS:

libraryDependencies += javaWs

Теперь мы можем запустить сервер с помощью команды sbt run :

$ sbt run
...
--- (Running the application, auto-reloading is enabled) ---

[info] p.c.s.AkkaHttpServer - Listening for HTTP on /0:0:0:0:0:0:0:0:9000

После запуска приложения мы можем проверить, все ли в порядке, просмотрев http://localhost:9000 , который откроет страницу приветствия Play.

3.2. Среда Тестирования

Для тестирования нашего приложения мы будем использовать класс модульного тестирования HomeControllerTest .

Во-первых, нам нужно расширить С помощью сервера , который обеспечит жизненный цикл сервера:

public class HomeControllerTest extends WithServer {

Благодаря своему родителю, этот класс теперь запускает наш скелетный веб-сервер в тестовом режиме и на случайном порту , прежде чем запускать тесты. Класс With Server также останавливает приложение по завершении теста.

Затем нам нужно предоставить приложение для запуска.

Мы можем создать его с помощью Guice ‘s GuiceApplicationBuilder :

@Override
protected Application provideApplication() {
    return new GuiceApplicationBuilder().build();
}

И, наконец, мы настроили URL-адрес сервера для использования в наших тестах, используя номер порта, предоставленный тестовым сервером:

@Override
@Before
public void setup() {
    OptionalInt optHttpsPort = testServer.getRunningHttpsPort();
    if (optHttpsPort.isPresent()) {
        port = optHttpsPort.getAsInt();
        url = "https://localhost:" + port;
    } else {
        port = testServer.getRunningHttpPort()
          .getAsInt();
        url = "http://localhost:" + port;
    }
}

Теперь мы готовы писать тесты. Комплексная структура тестирования позволяет нам сосредоточиться на кодировании наших тестовых запросов.

4. Подготовьте запрос WSRequest

Давайте посмотрим, как мы можем запускать основные типы запросов, такие как GET или POST, и составные запросы на загрузку файлов.

4.1. Инициализация объекта WSRequest

Прежде всего, нам нужно получить экземпляр WS Client для настройки и инициализации наших запросов.

В реальном приложении мы можем получить клиент, автоматически настроенный с настройками по умолчанию, с помощью инъекции зависимостей:

@Autowired
WSClient ws;

Однако в нашем тестовом классе мы используем WS Test Client , доступный из Игровой тестовый фреймворк :

WSClient ws = play.test.WSTestClient.newClient(port);

Как только у нас есть клиент, мы можем инициализировать объект WSRequest , вызвав метод url :

ws.url(url)

Метод url делает достаточно, чтобы мы могли запустить запрос. Тем не менее, мы можем настроить его дополнительно, добавив некоторые пользовательские настройки:

ws.url(url)
  .addHeader("key", "value")
  .addQueryParameter("num", "" + num);

Как мы видим, добавить заголовки и параметры запроса довольно легко.

После того, как мы полностью настроили наш запрос, мы можем вызвать метод, чтобы инициировать его.

4.2. Общий запрос на ПОЛУЧЕНИЕ

Чтобы вызвать запрос GET, нам просто нужно вызвать метод get для нашего объекта WSRequest :

ws.url(url)
  ...
  .get();

Поскольку это неблокирующий код, он запускает запрос, а затем продолжает выполнение в следующей строке нашей функции.

Объект , возвращаемый get , является Этапом завершения экземпляром , который является частью CompletableFuture API .

После завершения HTTP-вызова на этом этапе выполняется всего несколько инструкций. Он обертывает ответ в объект Response .

Обычно этот результат передается на следующий этап цепочки выполнения. В этом примере мы не предоставили никакой функции потребления, поэтому результат теряется.

По этой причине этот запрос относится к типу “огонь и забудь”.

4.3. Отправить форму

Отправка формы не сильно отличается от примера get .

Чтобы вызвать запрос, мы просто вызываем метод post :

ws.url(url)
  ...
  .setContentType("application/x-www-form-urlencoded")
  .post("key1=value1&key2=value2");

В этом сценарии нам нужно передать тело в качестве параметра. Это может быть простая строка, такая как файл, документ json или xml, Тело, доступное для записи или Источник .

4.4. Отправка данных из нескольких частей/Форм

Составная форма требует, чтобы мы отправляли как поля ввода, так и данные из вложенного файла или потока.

Чтобы реализовать это в рамках, мы используем пост метод с Источник .

Внутри источника мы можем обернуть все различные типы данных, необходимые для нашей формы:

Source file = FileIO.fromPath(Paths.get("hello.txt"));
FilePart> file = 
  new FilePart<>("fileParam", "myfile.txt", "text/plain", file);
DataPart data = new DataPart("key", "value");

ws.url(url)
...
  .post(Source.from(Arrays.asList(file, data)));

Хотя этот подход добавляет еще немного конфигурации, он все еще очень похож на другие типы запросов.

5. Обработайте асинхронный ответ

До этого момента мы запускали только запросы “огонь и забудь”, когда наш код ничего не делал с данными ответа.

Теперь давайте рассмотрим два метода обработки асинхронного ответа.

Мы можем либо заблокировать основной поток, ожидая CompletableFuture, или потреблять асинхронно с Потребитель .

5.1. Ответ процесса путем блокировки С помощью CompletableFuture

Даже при использовании асинхронной платформы мы можем заблокировать выполнение нашего кода и дождаться ответа.

Использование CompletableFuture API, нам просто нужно несколько изменений в нашем коде, чтобы реализовать этот сценарий:

WSResponse response = ws.url(url)
  .get()
  .toCompletableFuture()
  .get();

Это может быть полезно, например, для обеспечения надежной согласованности данных, которую мы не можем достичь другими способами.

5.2. Асинхронный Отклик Процесса

Чтобы обработать асинхронный ответ без блокировки, мы предоставляем Потребителя или Функцию , которая запускается асинхронной платформой, когда ответ доступен.

Например, давайте добавим Consumer в наш предыдущий пример для регистрации ответа:

ws.url(url)
  .addHeader("key", "value")
  .addQueryParameter("num", "" + 1)
  .get()
  .thenAccept(r -> 
    log.debug("Thread#" + Thread.currentThread().getId() 
      + " Request complete: Response code = " + r.getStatus() 
      + " | Response: " + r.getBody() 
      + " | Current Time:" + System.currentTimeMillis()));

Затем мы видим ответ в журналах:

[debug] c.HomeControllerTest - Thread#30 Request complete: Response code = 200 | Response: {
  "Result" : "ok",
  "Params" : {
    "num" : [ "1" ]
  },
  "Headers" : {
    "accept" : [ "*/*" ],
    "host" : [ "localhost:19001" ],
    "key" : [ "value" ],
    "user-agent" : [ "AHC/2.1" ]
  }
} | Current Time:1579303109613

Стоит отметить , что мы использовали , а затем приняли , для чего требуется функция Consumer , поскольку нам не нужно ничего возвращать после входа в систему.

Когда мы хотим, чтобы текущий этап вернул что-то, чтобы мы могли использовать это на следующем этапе, нам нужно , а затем применить вместо этого, который принимает функцию /.

Они используют соглашения стандартных функциональных интерфейсов Java .

5.3. Большой Орган Реагирования

Код, который мы реализовали до сих пор, является хорошим решением для небольших ответов и большинства вариантов использования. Однако, если нам нужно обработать несколько сотен мегабайт данных, нам понадобится лучшая стратегия.

Мы должны отметить: Методы запроса, такие как get и post , загружают весь ответ в память.

Чтобы избежать возможной ошибки OutOfMemoryError , мы можем использовать потоки Akka для обработки ответа, не позволяя ему заполнить нашу память.

Например, мы можем записать его тело в файл:

ws.url(url)
  .stream()
  .thenAccept(
    response -> {
        try {
            OutputStream outputStream = Files.newOutputStream(path);
            Sink> outputWriter =
              Sink.foreach(bytes -> outputStream.write(bytes.toArray()));
            response.getBodyAsSource().runWith(outputWriter, materializer);
        } catch (IOException e) {
            log.error("An error happened while opening the output stream", e);
        }
    });

Метод stream возвращает Стадию завершения , где ответ WS имеет getBodyAsStream метод, который предоставляет Source ?> . ?>

Мы можем рассказать коду , как обрабатывать этот тип тела, используя Sink Akka, который в нашем примере просто запишет любые данные, проходящие через OutputStream .

5.4. Тайм-ауты

При построении запроса мы также можем установить определенный тайм-аут, поэтому запрос прерывается, если мы не получим полный ответ вовремя.

Это особенно полезная функция, когда мы видим, что служба, к которой мы обращаемся, работает особенно медленно и может привести к накоплению открытых соединений, застрявших в ожидании ответа.

Мы можем установить глобальный тайм-аут для всех наших запросов, используя параметры настройки. Для тайм-аута конкретного запроса мы можем добавить к запросу с помощью setRequestTimeout :

ws.url(url)
  .setRequestTimeout(Duration.of(1, SECONDS));

Однако есть еще один случай, с которым нужно справиться: мы, возможно, получили все данные, но наш Потребитель может очень медленно обрабатывать их. Это может произойти, если есть много хруста данных, вызовов базы данных и т. Д.

В системах с низкой пропускной способностью мы можем просто позволить коду работать до тех пор, пока он не завершится. Однако мы, возможно, пожелаем прервать длительную деятельность.

Чтобы достичь этого, мы должны обернуть ваш код некоторыми фьючерсы обращение.

Давайте смоделируем очень длительный процесс в нашем коде:

ws.url(url)
  .get()
  .thenApply(
    result -> { 
        try { 
            Thread.sleep(10000L); 
            return Results.ok(); 
        } catch (InterruptedException e) { 
            return Results.status(SERVICE_UNAVAILABLE); 
        } 
    });

Это вернет ответ OK через 10 секунд, но мы не хотим ждать так долго.

Вместо этого с помощью обертки timeout мы инструктируем наш код ждать не более 1 секунды:

CompletionStage f = futures.timeout(
  ws.url(url)
    .get()
    .thenApply(result -> {
        try {
            Thread.sleep(10000L);
            return Results.ok();
        } catch (InterruptedException e) {
            return Results.status(SERVICE_UNAVAILABLE);
        }
    }), 1L, TimeUnit.SECONDS);

Теперь наше будущее вернет результат в любом случае: результат вычисления, если Потребитель закончил вовремя, или исключение из-за фьючерсов тайм-аута.

5.5. Обработка Исключений

В предыдущем примере мы создали функцию, которая либо возвращает результат, либо завершается ошибкой с исключением. Итак, теперь нам нужно обработать оба сценария.

Мы можем справиться как со сценариями успеха, так и со сценариями неудачи с помощью обрабатывать асинхронность метод.

Допустим, мы хотим вернуть результат, если он у нас есть, или зарегистрировать ошибку и вернуть исключение для дальнейшей обработки:

CompletionStage

Теперь код должен возвращать Стадию завершения , содержащую Исключение TimeoutException .

Мы можем проверить это, просто вызвав assertEquals в классе возвращаемого объекта исключения:

Class clazz = res.toCompletableFuture().get().getClass();
assertEquals(TimeoutException.class, clazz);

При запуске теста он также зарегистрирует полученное нами исключение:

[error] c.HomeControllerTest - Exception thrown
java.util.concurrent.TimeoutException: Timeout after 1 second
...

6. Фильтры запросов

Иногда нам нужно выполнить некоторую логику, прежде чем будет запущен запрос.

Мы могли бы манипулировать объектом WSRequest после инициализации, но более элегантным методом является установка Warequestfilter .

Фильтр может быть установлен во время инициализации, перед вызовом метода запуска, и присоединен к логике запроса.

Мы можем определить наш собственный фильтр, реализовав интерфейс Warequestfilter , или мы можем добавить готовый фильтр.

Распространенным сценарием является регистрация того, как выглядит запрос, перед его выполнением.

В этом случае нам просто нужно установить AhcCurlRequestLogger :

ws.url(url)
  ...
  .setRequestFilter(new AhcCurlRequestLogger())
  ...
  .get();

Результирующий журнал имеет формат curl -like:

[info] p.l.w.a.AhcCurlRequestLogger - curl \
  --verbose \
  --request GET \
  --header 'key: value' \
  'http://localhost:19001'

Мы можем установить желаемый уровень журнала, изменив ваш logback.xml конфигурация.

7. Кэширование Ответов

WSClient также поддерживается кэширование ответов.

Эта функция особенно полезна, когда один и тот же запрос запускается несколько раз, и нам не нужны самые свежие данные каждый раз.

Это также помогает, когда служба, которую мы вызываем, временно отключена.

7.1. Добавление Зависимостей Кэширования

Чтобы настроить кэширование, нам нужно сначала добавить зависимость в наш build.sbt :

libraryDependencies += ehcache

Это настраивает Ehcache в качестве нашего слоя кэширования.

Если мы не хотим специально использовать Ehcache, мы можем использовать любую другую реализацию кэша JSR-107 .

7.2. Эвристика Принудительного Кэширования

По умолчанию Play WS не будет кэшировать HTTP-ответы, если сервер не возвращает никакой конфигурации кэширования.

Чтобы обойти это, мы можем принудительно выполнить эвристическое кэширование, добавив параметр в наш application.conf :

play.ws.cache.heuristics.enabled=true

Это позволит настроить систему так, чтобы она решала, когда полезно кэшировать HTTP-ответ, независимо от объявленного кэширования удаленной службы.

8. Дополнительная Настройка

Для выполнения запросов к внешней службе может потребоваться некоторая конфигурация клиента. Возможно, нам придется обрабатывать перенаправления, медленный сервер или некоторую фильтрацию в зависимости от заголовка агента пользователя.

Чтобы решить эту проблему, мы можем настроить наш клиент WS, используя свойства в нашем Чтобы решить эту проблему, мы можем настроить наш клиент WS, используя свойства в нашем :

play.ws.followRedirects=false
play.ws.useragent=MyPlayApplication
play.ws.compressionEnabled=true
# time to wait for the connection to be established
play.ws.timeout.connection=30
# time to wait for data after the connection is open
play.ws.timeout.idle=30
# max time available to complete the request
play.ws.timeout.request=300

Также можно настроить базовую AsyncHttpClient непосредственно.

Полный список доступных свойств можно проверить в исходном коде Acconfig .

9. Заключение

В этой статье мы рассмотрели библиотеку Play WS и ее основные функции. Мы настроили наш проект, научились запускать общие запросы и обрабатывать их ответы как синхронно, так и асинхронно.

Мы работали с большими загрузками данных и видели, как сократить длительные действия.

Наконец, мы рассмотрели кэширование для повышения производительности и способы настройки клиента.

Как всегда, исходный код этого учебника доступен на GitHub .