Используешь ли очереди

Используешь ли очереди?

Да, очереди — это один из основных инструментов в Java для управления асинхронной обработкой и межпоточной коммуникацией. Я активно использую их в различных сценариях.

1. BlockingQueue для многопоточных приложений

Сценарий: Обработка задач от нескольких потоков производителя и несколько потоков потребителя.

import java.util.concurrent.*;

public class TaskProcessor {
    private final BlockingQueue<Task> taskQueue = new LinkedBlockingQueue<>(100);
    private final ExecutorService executor = Executors.newFixedThreadPool(5);
    
    public void submitTask(Task task) throws InterruptedException {
        taskQueue.put(task);  // Блокирует если очередь переполнена
    }
    
    public void startProcessing() {
        for (int i = 0; i < 5; i++) {
            executor.submit(() -> {
                while (true) {
                    try {
                        Task task = taskQueue.take();  // Блокирует если очередь пуста
                        processTask(task);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            });
        }
    }
    
    private void processTask(Task task) {
        // Обработка задачи
        System.out.println("Обработка: " + task);
    }
}

// Использование
public static void main(String[] args) throws InterruptedException {
    TaskProcessor processor = new TaskProcessor();
    processor.startProcessing();
    
    // Производители
    for (int i = 0; i < 10; i++) {
        processor.submitTask(new Task("task-" + i));
    }
}

2. PriorityQueue для приоритизации задач

Сценарий: Важные задачи должны обрабатываться раньше обычных.

import java.util.*;
import java.util.concurrent.*;

public class PriorityTaskQueue {
    private final PriorityBlockingQueue<PrioritizedTask> queue = 
        new PriorityBlockingQueue<>();
    
    public void addTask(String name, int priority) {
        queue.offer(new PrioritizedTask(name, priority));
    }
    
    public void processTasks() {
        while (!queue.isEmpty()) {
            PrioritizedTask task = queue.poll();
            System.out.println("Обработка: " + task);
        }
    }
    
    static class PrioritizedTask implements Comparable<PrioritizedTask> {
        String name;
        int priority;
        
        PrioritizedTask(String name, int priority) {
            this.name = name;
            this.priority = priority;
        }
        
        @Override
        public int compareTo(PrioritizedTask other) {
            return Integer.compare(other.priority, this.priority);  // Обратный порядок
        }
        
        @Override
        public String toString() {
            return name + " [priority: " + priority + "]";
        }
    }
}

// Использование
PriorityTaskQueue queue = new PriorityTaskQueue();
queue.addTask("backup", 1);
queue.addTask("critical-alert", 10);
queue.addTask("log-cleanup", 2);
queue.processTasks();
// Output:
// Обработка: critical-alert [priority: 10]
// Обработка: log-cleanup [priority: 2]
// Обработка: backup [priority: 1]

3. ConcurrentLinkedQueue для высоконагруженных систем

Сценарий: Очередь без блокировок для максимальной производительности.

import java.util.concurrent.*;

public class HighPerformanceQueue {
    private final ConcurrentLinkedQueue<Event> events = new ConcurrentLinkedQueue<>();
    
    public void publishEvent(Event event) {
        events.offer(event);  // Non-blocking, O(1) операция
    }
    
    public void processEvents() {
        while (true) {
            Event event = events.poll();
            if (event == null) {
                // Очередь пуста
                Thread.sleep(100);
                continue;
            }
            handleEvent(event);
        }
    }
    
    private void handleEvent(Event event) {
        System.out.println("Event: " + event);
    }
}

4. DelayQueue для отложенных операций

Сценарий: Выполнение задач с задержкой (кэш-экспирация, таймауты).

import java.util.concurrent.*;

public class CacheWithExpiration {
    private final DelayQueue<CacheEntry> expiredEntries = new DelayQueue<>();
    private final Map<String, String> cache = new ConcurrentHashMap<>();
    
    public void put(String key, String value, long expireMs) {
        cache.put(key, value);
        expiredEntries.offer(new CacheEntry(key, expireMs));
    }
    
    public void cleanupExpired() {
        new Thread(() -> {
            while (true) {
                try {
                    CacheEntry expired = expiredEntries.take();  // Ждет пока элемент задержится
                    cache.remove(expired.key);
                    System.out.println("Удален из кэша: " + expired.key);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }).start();
    }
    
    static class CacheEntry implements Delayed {
        String key;
        long expireTime;
        
        CacheEntry(String key, long delayMs) {
            this.key = key;
            this.expireTime = System.currentTimeMillis() + delayMs;
        }
        
        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(expireTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }
        
        @Override
        public int compareTo(Delayed other) {
            return Long.compare(this.expireTime, ((CacheEntry) other).expireTime);
        }
    }
}

5. LinkedBlockingDeque для двусторонней очереди

Сценарий: Обработка с возможностью добавления с обоих концов (работа воровства).

import java.util.concurrent.*;

public class WorkStealingExample {
    private final LinkedBlockingDeque<Task> workDeque = new LinkedBlockingDeque<>();
    
    // Основной поток добавляет в конец
    public void addWork(Task task) throws InterruptedException {
        workDeque.putLast(task);
    }
    
    // Рабочий поток берет из конца (его собственная работа)
    public Task getOwnWork() throws InterruptedException {
        return workDeque.pollLast(1, TimeUnit.SECONDS);
    }
    
    // Обиженный рабочий может украсть работу спереди (от другого)
    public Task stealWork() throws InterruptedException {
        return workDeque.pollFirst(1, TimeUnit.SECONDS);
    }
}

6. Queue для обработки сообщений в приложениях

Сценарий: Система обработки заказов e-commerce.

public class OrderProcessingSystem {
    private final BlockingQueue<Order> orderQueue = new LinkedBlockingQueue<>();
    private final ExecutorService executors = Executors.newFixedThreadPool(10);
    
    public OrderProcessingSystem() {
        startOrderProcessors();
    }
    
    public void submitOrder(Order order) throws InterruptedException {
        orderQueue.put(order);
    }
    
    private void startOrderProcessors() {
        for (int i = 0; i < 10; i++) {
            executors.submit(() -> {
                while (!Thread.currentThread().isInterrupted()) {
                    try {
                        Order order = orderQueue.take();
                        processOrder(order);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            });
        }
    }
    
    private void processOrder(Order order) {
        System.out.println("Processing order: " + order.id);
        // Проверка запасов
        // Обработка платежа
        // Отправка уведомления
        // Создание отправки
    }
    
    public static class Order {
        long id;
        String customer;
        
        Order(long id, String customer) {
            this.id = id;
            this.customer = customer;
        }
    }
}

7. Сравнение различных типов очередей

Рекомендации по использованию

1. Выбирай BlockingQueue для стандартной многопоточной обработки
2. Используй PriorityBlockingQueue для приоритизации
3. ConcurrentLinkedQueue для максимальной производительности без блокировок
4. Избегай busy-waiting — используй take() вместо poll() в цикле
5. Всегда обрабатывай InterruptedException правильно
6. Используй ExecutorService вместо ручного управления потоками

Итоги

Очереди — это фундаментальный инструмент для:

• Асинхронной обработки
• Балансировки нагрузки
• Межпоточной коммуникации
• Обработки сообщений
• Управления ресурсами

Выбор правильного типа очереди критичен для производительности и надежности приложения.