Полное руководство по асинхронному программированию на SpringBoot для новичков

@FunctionalInterface
public interface TaskExecutor extends Executor {
    void execute(Runnable var1);
}

**Основные понятия `ThreadPoolTaskExecutor`:**- **Core Pool Size :** Количество ядерных потоков определяет минимальное количество одновременно выполняемых потоков.
- **Queue Capacity :** Когда поступает новая задача, сначала проверяется, достигло ли количество текущих потоков значения ядерных потоков. Если достигнуто, задача будет помещена в очередь.
- **Maximum Pool Size :** Когда количество задач в очереди достигает максимального размера очереди, количество одновременно выполняемых потоков становится максимальным. **Что произойдет, если очередь уже заполнена, а количество одновременно выполняющихся потоков достигло максимального значения, и поступит новая задача?**По умолчанию Spring использует `ThreadPoolExecutor.AbortPolicy`. В этом случае `ThreadPoolExecutor` выбросит исключение `RejectedExecutionException`, отклонив новую задачу. Это означает, что обработка этой задачи будет потеряна. Для масштабируемых приложений рекомендуется использовать `ThreadPoolExecutor.CallerRunsPolicy`. Эта стратегия выполняет задачу в том потоке, который её отправил, что обеспечивает масштабируемость при заполненной максимальной очереди.

**Определение стратегии насыщения `ThreadPoolTaskExecutor`:**

Если количество одновременно выполняющихся потоков достигло максимального значения, `ThreadPoolTaskExecutor` определяет следующие стратегии:

- **ThreadPoolExecutor.AbortPolicy**: Выбрасывает исключение `RejectedExecutionException`, чтобы отклонить обработку новой задачи.
- **ThreadPoolExecutor.CallerRunsPolicy**: Задача выполняется в том потоке, который её отправил. Вы не потеряете запрос на задачу. Однако эта стратегия может замедлить скорость подачи новых задач, что может негативно сказаться на производительности программы. Кроме того, эта стратегия предпочитает увеличение размера очереди. Если ваше приложение может выдержать эту задержку и вы не можете позволить себе потерю даже одного запроса на задачу, вы можете выбрать эту стратегию.
- **ThreadPoolExecutor.DiscardPolicy**: Новая задача не обрабатывается и просто отбрасывается.
- **ThreadPoolExecutor.DiscardOldestPolicy**: Эта стратегия отбрасывает самую старую необработанную задачу.### 2. Написание асинхронного метода

Ниже представлен пример метода, который ищет фильмы, начинающиеся с определённой буквы. Мы добавили аннотацию `@Async`, чтобы указать Spring, что это асинхронный метод. Возвращаемое значение метода `CompletableFuture.completedFuture(results)` означает, что нам нужно вернуть результат, то есть программа должна завершить выполнение задачи перед тем, как вернуть результат пользователю.

**Обратите внимание на первую строку лога в методе `completableFutureTask`, эта строка будет использована в анализе программы и является очень важной!**

```java
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

/** @author shuang.kou */
@Service
public class AsyncService {
  
  private static final Logger logger = LoggerFactory.getLogger(AsyncService.class);

  private List<String> movies =
      new ArrayList<>(Arrays.asList(
          "Forrest Gump",
          "Titanic",
          "Spirited Away",
          "The Shawshank Redemption",
          "Zootopia",
          "Farewell",
          "Joker",
          "Crawl"));
```  /**
   * Пример использования: найти фильмы, начинающиеся с определенного символа/строки
   */
  @Async
  public CompletableFuture<List<String>> completableFutureTask(String start) {
    // Вывод лога
    logger.warn(Thread.currentThread().getName() + " начал выполнение задачи!");
    // Найти фильмы, начинающиеся с определенного символа/строки
    List<String> results =
        movies.stream().filter(movie -> movie.startsWith(start)).collect(Collectors.toList());
    // Моделирование долгой задачи
    try {
      Thread.sleep(1000L);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    // Возвращение нового CompletableFuture, завершенного заданным значением
    return CompletableFuture.completedFuture(results);
  }
}```

### 3. Тестирование написанного асинхронного метода

```java
/** @author shuang.kou */
@RestController
@RequestMapping("/async")
public class AsyncController {
  @Autowired
  AsyncService asyncService;

  @GetMapping("/movies")
  public String completableFutureTask() throws ExecutionException, InterruptedException {
    // Время начала
    long start = System.currentTimeMillis();
    // Начало выполнения большого количества асинхронных задач
    List<String> words = Arrays.asList("F", "T", "S", "Z", "J", "C");
    List<CompletableFuture<List<String>>> completableFutureList =
        words.stream()
            .map(word -> asyncService.completableFutureTask(word))
            .collect(Collectors.toList());
    // Метод CompletableFuture.join() позволяет получить результаты и объединить их
    List<List<String>> results = completableFutureList.stream().map(CompletableFuture::join).collect(Collectors.toList());
    // Вывод результата и времени выполнения программы
    System.out.println("Время выполнения: " + (System.currentTimeMillis() - start));
    return results.toString();
  }
}

Сначала мы можем заметить, что время, затраченное на выполнение всех задач, составляет примерно 1 с. Это связано с нашим пользовательским `ThreadPoolTaskExecutor`, в котором мы настроили количество ядерных потоков на 6, а затем с помощью следующего кода симулировали распределение 6 задач для выполнения системой. Таким образом, каждый поток получает по одной задаче, и время выполнения каждой задачи составляет 1 с, поэтому общее время выполнения 6 задач составляет 1 с.```java
List<String> words = Arrays.asList("F", "T", "S", "Z", "J", "C");  
List<CompletableFuture<Void>> completableFutureList =
        words.stream()
            .map(word -> asyncService.completableFutureTask(word))
            .collect(Collectors.toList());

@Async
public void completableFutureTask(String start) {
  ......
  // Здесь может быть обработка результата выполнения задачи, например, запись в базу данных и т.д.
  // doSomeThingWithResults(results);
}
```Измените код контроллера следующим образом:
```java
  @GetMapping("/movies")
  public String completableFutureTask() throws ExecutionException, InterruptedException {
    // Запуск таймера
    long start = System.currentTimeMillis();
    // Запуск нескольких асинхронных поисков
    List<String> words = Arrays.asList("F", "T", "S", "Z", "J", "C");
    words.stream()
        .forEach(word -> asyncService.completableFutureTask(word));
    // Ожидание завершения всех задач
    // Вывод результатов, включая затраченное время
    System.out.println("Затраченное время: " + (System.currentTimeMillis() - start));
    return "Завершено";
  }

Затраченное время: 0
2019-10-01 14:02:44.052  WARN 19051 --- [lTaskExecutor-4] g.j.a.service.AsyncService               : My ThreadPoolTaskExecutor-4start this task!
2019-10-01 14:02:44.052  WARN 19051 --- [lTaskExecutor-3] g.j.a.service.AsyncService               : My ThreadPoolTaskExecutor-3start this task!
2019-10-01 14:02:44.052  WARN 19051 --- [lTaskExecutor-2] g.j.a.service.AsyncService               : My ThreadPoolTaskExecutor-2start this task!
2019-10-01 14:02:44.052  WARN 19051 --- [lTaskExecutor-1] g.j.a.service.AsyncService               : My ThreadPoolTaskExecutor-1start this task!
2019-10-01 14:02:44.052  WARN 19051 --- [lTaskExecutor-6] g.j.a.service.AsyncService               : My ThreadPoolTaskExecutor-6start this task!
2019-10-01 14:02:44.052  WARN 19051 --- [lTaskExecutor-5] g.j.a.service.AsyncService               : My ThreadPoolTaskExecutor-5start this task!