Очереди задач в Java: практическое руководство с примерами

Онлайн-тренажер Python для начинающих

Изучайте Python легко и без перегрузки теорией. Решайте практические задачи с автоматической проверкой, получайте подсказки на русском языке и пишите код прямо в браузере — без необходимости что-либо устанавливать.

Начать курс

Введение: зачем нужны очереди задач в Java?

В современной разработке на Java многопоточность — это не роскошь, а необходимость. Приложения обрабатывают тысячи запросов, выполняют фоновые задачи и взаимодействуют с внешними сервисами. Без правильной организации очередей задач код превращается в хаос: потоки блокируют друг друга, память переполняется, а производительность падает.

Очереди задач (Task Queues) — это механизм, который позволяет эффективно распределять работу между потоками. В Java для этого существует мощный набор инструментов в пакете java.util.concurrent. В этой статье мы разберем, как использовать очереди задач на практике, с реальными примерами кода.

Вы узнаете:

Как работают BlockingQueue и её реализации
Как настроить ThreadPoolExecutor для конкретных задач
Как избежать типичных ошибок при работе с очередями
Как тестировать и отлаживать многопоточный код

Основы очередей задач: BlockingQueue

BlockingQueue — это интерфейс, который лежит в основе большинства очередей задач в Java. Он поддерживает операции ожидания: если очередь пуста, поток-потребитель ждет, пока появится элемент; если очередь переполнена, поток-производитель ждет, пока освободится место.

Основные реализации BlockingQueue

В Java есть несколько реализаций BlockingQueue, каждая со своими особенностями:

ArrayBlockingQueue — фиксированный размер, основана на массиве. Подходит для сценариев с известным лимитом задач.
LinkedBlockingQueue — опционально ограниченная, основана на связном списке. Хороша для динамических нагрузок.
PriorityBlockingQueue — неограниченная очередь с приоритетами. Используется, когда задачи имеют разную важность.
SynchronousQueue — очередь без容量: каждый insert должен ждать соответствующего remove. Идеально для прямой передачи задач.

Пример: простая очередь задач с ArrayBlockingQueue

Рассмотрим базовый пример, где один поток генерирует задачи, а другой их обрабатывает:

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class SimpleTaskQueue {
    public static void main(String[] args) {
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(10);

        // Поток-производитель
        Thread producer = new Thread(() -> {
            try {
                for (int i = 1; i <= 20; i++) {
                    String task = "Задача №" + i;
                    queue.put(task); // Блокируется, если очередь полна
                    System.out.println("Добавлена: " + task);
                    Thread.sleep(100); // Имитация генерации
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        // Поток-потребитель
        Thread consumer = new Thread(() -> {
            try {
                while (true) {
                    String task = queue.take(); // Блокируется, если очередь пуста
                    System.out.println("Обработана: " + task);
                    Thread.sleep(300); // Имитация обработки
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });

        producer.start();
        consumer.start();
    }
}

Этот код демонстрирует ключевой паттерн: производитель добавляет задачи через put(), потребитель забирает через take(). Оба метода блокируются при необходимости, что обеспечивает естественную синхронизацию.

ThreadPoolExecutor: профессиональная очередь задач

На практике редко используют сырые очереди напрямую. Вместо этого применяют ThreadPoolExecutor — мощный фреймворк для управления пулом потоков и очередью задач.

Как настроить ThreadPoolExecutor

Конструктор ThreadPoolExecutor принимает несколько параметров:

corePoolSize — минимальное количество потоков, которые всегда активны
maximumPoolSize — максимальное количество потоков
keepAliveTime — время простоя, после которого лишние потоки уничтожаются
workQueue — очередь для хранения ожидающих задач
handler — стратегия, когда очередь переполнена и все потоки заняты

Пример: настраиваемый пул потоков

import java.util.concurrent.*;

public class CustomThreadPool {
    public static void main(String[] args) {
        // Очередь с лимитом в 5 задач
        BlockingQueue<Runnable> taskQueue = new ArrayBlockingQueue<>(5);

        // Стратегия: отклонить задачу и выбросить RejectedExecutionException
        RejectedExecutionHandler rejectionHandler = new ThreadPoolExecutor.AbortPolicy();

        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            2,       // corePoolSize
            4,       // maximumPoolSize
            60,      // keepAliveTime
            TimeUnit.SECONDS,
            taskQueue,
            rejectionHandler
        );

        // Отправляем 10 задач
        for (int i = 1; i <= 10; i++) {
            final int taskId = i;
            executor.submit(() -> {
                System.out.println("Выполняется задача " + taskId + 
                                   " потоком " + Thread.currentThread().getName());
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }

        executor.shutdown();
    }
}

В этом примере: 2 потока работают постоянно, при росте нагрузки создаются до 4 потоков. Если все 4 заняты и очередь из 5 задач переполнена, новые задачи отклоняются. Это предотвращает перегрузку системы.

Готовые фабрики: Executors

Для типовых сценариев Java предлагает фабричные методы в классе Executors:

// Фиксированный пул из 4 потоков
ExecutorService fixedPool = Executors.newFixedThreadPool(4);

// Кэшируемый пул (потоки создаются по мере необходимости)
ExecutorService cachedPool = Executors.newCachedThreadPool();

// Пул с одним потоком (гарантирует последовательное выполнение)
ExecutorService singlePool = Executors.newSingleThreadExecutor();

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