Исходные данные

Java 8 и выше

Spring 4.1 и выше

Для чего необходимо асинхронное выполнение (асинхронный вызов, async invocation)?

Основных применений несколько:

1. Улучшение производительности приложения

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

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

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

На иллюстрации ниже схематично показано различие между синхронным и асинхронным выполнением программы.

2. Выделение затратных (по времени и ресурсам - expensive jobs) задач в отдельные потоки и/или выполнение их в фоновом режиме

Рассмотрим пример. Предположим, что мы разрабатываем приложение, в котором реализуем оплату заказа. Сам процесс выполнения и зачисления оплаты занимает некоторое продолжительное время. Мы хотим, чтобы пользователь не ждал окончания всего процесса и мог продолжить пользоваться приложением. Результат оплаты пользователь сможет в дальнейшем посмотреть в своем личном кабинете или, например, получив уведомление на электронную почту или телефон.

Чтобы осуществить описанный выше сценарий, нам необходимо инициировать процесс оплаты асинхронно в отдельном потоке.

Добавление поддержки асинхронности в приложение

Аннотация @EnableAsync

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

import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.AsyncConfigurerSupport;
import org.springframework.scheduling.annotation.EnableAsync;


@Configuration
@EnableAsync
public class AsyncConfiguration extends AsyncConfigurerSupport {

//…. код опущен для краткости

}

Аннотация @EnableAsync включает следующее поведение Spring:

- Spring распознает методы, помеченные аннотацией @Async

- Spring запускает эти методы в фоновом пуле потоков (background thread pool)

Аннотация @Async

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

@Async
public void asyncMethod(SomeDomainClass someDomainObject) {
 //долго выполняющийся фоновый процесс
}

Существует несколько правил, касательно данной аннотации:

А) Аннотация @Async должна применяться только на публичных методах (public). Дело в том, что Spring создает прокси для метода с этой аннотацией, и для работы прокси метод должен быть публичным.

Б) Нельзя вызывать метод, помеченный @Async из того же класса, где он определен. Такой вызов не сработает, так как он будет обходить прокси (proxy bypass).

В) Если метод, помеченный @Async должен что-то возвращать, то его возвращаемый тип необходимо определить как CompletableFuture или Future:

@Async
public CompletableFuture<SomeDomainClass> getObjectById(final Long id) throws InterruptedException {
SomeDomainClass someDomainObject = new SomeDomainClass();

    
//… процесс получения данных и их обработки опущен для краткости


    return CompletableFuture.completedFuture(someDomainObject);
}

Исполнитель задач (The Task Executor)

Spring использует пул потоков для управления потоками фоновых процессов. В свою очередь, для управления и конфигурации пула потоков Spring использует Исполнитель задач - бин TaskExecutor.

По умолчанию аннотация @EnableAsync создает SimpleAsyncTaskExecutor.

К сожалению, эта реализация не имеет фактического верхнего предела для размера пула потоков. Это означает, что приложение Spring может упасть (crash), если одновременно будет запущено слишком много методов с аннотацией @Async.

Чтобы избежать этого, нам необходимо переопределить собственный исполнитель задач.

Вернемся к нашему конфигурационному классу и добавим в него определение Исполнителя задач:

import java.util.concurrent.Executor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

@Configuration
@EnableAsync
public class AsyncConfiguration extends AsyncConfigurerSupport {

@Override
   public Executor getAsyncExecutor() {
      ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
      executor.setCorePoolSize(4);
      executor.setMaxPoolSize(10);
      executor.setQueueCapacity(50);
      executor.setThreadNamePrefix("AsyncTaskThread::");
      executor.setWaitForTasksToCompleteOnShutdown(true);
      executor.initialize();
      return executor;
  }

//…. Другой код опущен для краткости

}

Обработка ошибок

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

- Если метод возвращает CompletableFuture или Future, то вызов в нем Future.get() при наличии ошибок приведет к тому, что исключение (exception) будет выдано.

- Во всех остальных случаях (в особенности если метод ничего не возвращает – void), нам необходимо явно указать обработчик ошибок, так как по умолчанию исключения не будут передаваться вызывающему потоку.

Асинхронный обработчик ошибок должен реализовывать интерфейс AsyncUncaughtExceptionHandler.

Вновь возвращаемся к конфигурационному классу:

import java.lang.reflect.Method;
import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;

@Configuration
@EnableAsync
public class AsyncConfiguration extends AsyncConfigurerSupport {

@Override
   public Executor getAsyncExecutor() {
      ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
      executor.setCorePoolSize(4);
      executor.setMaxPoolSize(10);
      executor.setQueueCapacity(50);
      executor.setThreadNamePrefix("AsyncTaskThread::");
      executor.setWaitForTasksToCompleteOnShutdown(true);
      executor.initialize();
      return executor;
  }

@Override
  public AsyncUncaughtExceptionHandler  getAsyncUncaughtExceptionHandler() {
     return new AsyncUncaughtExceptionHandler() {
        @Override
        public void handleUncaughtException(Throwable ex, 
           Method method, Object... params) {

//Используйте здесь ваш любимый Logger
//Задавайте удобный для вас формат логирования

//Здесь для простоты примера использован метод System.out.println

           System.out.println("Exception: " + ex.getMessage());
           System.out.println("Method Name: " + method.getName());
           ex.printStackTrace();
        }
    };
  }}

Применение асинхронности в приложении

Для примера предположим, что мы разрабатываем часть API, которая ответственна за взаимодействие с данными пользователей, а именно – за получение данных о пользователях и за обновление (изменение) этих данных. Причем нам необходимо предусмотреть получение данных как в синхронном режиме, так и в асинхронном.

И так приступим.

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

Доменный класс, который описывает данные пользователя (или самого пользователя) пусть имеет следующий вид:

import lombok.Getter;
import lombok.Setter;

@Getter
@Setter
public class User {

    private Long id;
    private String firstName;
    private String lastName;
    private String middleName;
    private int age;
    private String email;
    private String username;

}

Напишем простой сервис для получения данных пользователей.

Для начала опишем интерфейс:

import java.util.concurrent.CompletableFuture;

public interface UserDataServiceCommon {

    CompletableFuture<User> getUserByEmail(final String email) throws InterruptedException;
    void updateUserData(User user) throws Exception;
    User getUserById(final Long userId) throws InterruptedException; 

}

Далее приступим к реализации сервиса:

import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import java.util.concurrent.CompletableFuture;

@Service
public class UserDataService implements UserDataServiceCommon{

//Асинхронный вызов
@Override
    @Async
    public CompletableFuture<User> getUserByEmail (final String email) throws InterruptedException {
  
         //Реализация получения данных пользователей опущена здесь для краткости
         //В данном примере используется паттерн Репозиторий для работы с источником данных
    
         User u = userRepository.findUserByEmail(email);

        return CompletableFuture.completedFuture(u);
    }

    @Override
    @Async
    public void updateUserData(User user) throws Exception {

         //Работа с данными и/или другие операции опущены для краткости 

       userRepository.save(user);

    }

//Синхронный вызов
    @Override
    public User getUserById (final Long userId) throws InterruptedException {

        User u = userRepository.findUserById(userId);

        return u;

    }

}

Далее напишем класс контроллера:

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.util.concurrent.CompletableFuture;

@RestController
@RequestMapping("/users")
public class UserController {

    @Autowired
    UserDataService uds;

    @GetMapping("/user/{email}")
    public CompletableFuture<User> getUserByEmail(@PathVariable String email) throws InterruptedException {

        return uds.findUserByEmail(email);

    }

    @PostMapping("/user/update-data")
    public void updateUserData() throws Exception {

        //Получение данных пользователя из тела post запроса опущено для краткости 
        User u = getUserFromRequestBody();
        uds.updateUserData(u);

    }

    @GetMapping("/user/id/{id}")
    public User getUserById (@PathVariable Long id) throws InterruptedException {

        return uds.getUserById(id);

    }
}

Заключение

В данной статье мы рассмотрели один из способов асинхронного программирования для Spring. Как мы увидели, асинхронность очень полезна для увеличения производительности приложения, для отделения долго выполняющихся процессов в отдельный пул потоков. По мимо этого, мы разобрали как использовать аннотации @EnableAsync и @Async, как переопределить Исполнитель задач, а также как написать обработчик ошибок (исключений).