Очереди задач в Java: от ThreadPoolExecutor до Disruptor

Очереди задач в Java: от встроенных механизмов до современных фреймворков

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

1. Встроенные механизмы: ThreadPoolExecutor и BlockingQueue

Стандартная библиотека Java (java.util.concurrent) предлагает мощный набор инструментов для работы с очередями задач. Основой является класс ThreadPoolExecutor, который управляет пулом потоков и использует очередь для хранения ожидающих задач.

Базовый пример:

import java.util.concurrent.*;

public class TaskQueueExample {
    public static void main(String[] args) {
        // Создаём пул потоков: corePoolSize=2, maxPoolSize=4, очередь на 100 задач
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
            2, // corePoolSize
            4, // maximumPoolSize
            60, // keepAliveTime
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(100) // очередь задач
        );

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

        // Корректное завершение
        executor.shutdown();
    }
}

Типы очередей для ThreadPoolExecutor:

• LinkedBlockingQueue — неограниченная очередь на основе связного списка. Подходит для систем, где количество задач заранее неизвестно.
• ArrayBlockingQueue — ограниченная очередь на основе массива. Позволяет контролировать максимальное количество ожидающих задач.
• SynchronousQueue — очередь без ёмкости: каждый put() ждёт take(). Используется для прямого проброса задач потокам.
• PriorityBlockingQueue — очередь с приоритетом. Задачи сортируются по заданному компаратору.

2. Архитектура Producer-Consumer на основе очередей

Классический паттерн Producer-Consumer легко реализуется с помощью BlockingQueue. Поставщики (producers) помещают задачи в очередь, а потребители (consumers) извлекают их для обработки.

Пример реализации:

import java.util.concurrent.*;

public class ProducerConsumerExample {
    private static final BlockingQueue<String> queue = new LinkedBlockingQueue<>(5);

    static class Producer implements Runnable {
        @Override
        public void run() {
            try {
                for (int i = 0; i < 10; i++) {
                    String task = "Задача-" + i;
                    queue.put(task); // блокируется, если очередь полна
                    System.out.println("Произведена: " + task);
                    Thread.sleep(100);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    static class Consumer implements Runnable {
        @Override
        public void run() {
            try {
                while (true) {
                    String task = queue.take(); // блокируется, если очередь пуста
                    System.out.println("Обработана: " + task + " [" + 
                                       Thread.currentThread().getName() + "]");
                    Thread.sleep(200);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(3);
        executor.submit(new Producer());
        executor.submit(new Consumer());
        executor.submit(new Consumer());
        executor.shutdown();
    }
}

3. Продвинутые решения: LMAX Disruptor

Для высоконагруженных систем (биржи, игровые серверы, обработка событий в реальном времени) стандартные очереди могут быть узким местом. LMAX Disruptor — это высокопроизводительная кольцевая очередь, которая обеспечивает минимальную задержку (latency) и максимальную пропускную способность (throughput).

Ключевые особенности Disruptor:

• Работает на основе кольцевого буфера (RingBuffer) фиксированного размера.
• Избегает блокировок (lock-free) с помощью CAS-операций.
• Поддерживает многопоточных производителей и потребителей.
• Гарантирует последовательную обработку событий.

Пример использования Disruptor (упрощённый):

import com.lmax.disrupt