Многопоточность в Java: кастомные пулы потоков и планировщики

// Старая школа: конкретный поток
class MyThread extends Thread {
    public void run() {
        System.out.println("Поток выполняется через наследование от Thread");
    }
}
 
// Гибкость: отделяем задачу от исполнителя
class MyRunnable implements Runnable {
    public void run() {
        System.out.println("Поток выполняется через Runnable");
    }
}
 
// Современный подход: задача с результатом
class MyCallable implements Callable<String> {
    public String call() throws Exception {
        return "Результат выполнения Callable";
    }
}

ExecutorService executor = Executors.newFixedThreadPool(5);
executor.submit(() -> {
    System.out.println("Задача выполняется в потоке " + Thread.currentThread().getName());
    return "Готово!";
});

ThreadPoolExecutor customPool = new ThreadPoolExecutor(
    4,                         // corePoolSize - постоянная бригада
    10,                        // maximumPoolSize - максимально возможное раширение
    60, TimeUnit.SECONDS,      // keepAliveTime - как долго "лишние" потоки ждут работы
    new LinkedBlockingQueue<Runnable>(100), // очередь задач
    new CustomThreadFactory(), // фабрика для создания потоков
    new CustomRejectionHandler() // что делать с непринятыми задачами
);

// Пример с CallerRunsPolicy
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    2, 2, 0, TimeUnit.MILLISECONDS,
    new LinkedBlockingQueue<Runnable>(10),
    Executors.defaultThreadFactory(),
    new ThreadPoolExecutor.CallerRunsPolicy()
);
 
// Пример собственной политики отклонения
ThreadPoolExecutor executor = new ThreadPoolExecutor(
    2, 2, 0, TimeUnit.MILLISECONDS,
    new LinkedBlockingQueue<Runnable>(10),
    Executors.defaultThreadFactory(),
    (r, executor) -> {
        System.err.println("Задача отклонена, логирую и пытаюсь ещё раз позже");
        // Логирование, метрики, повторные попытки...
    }
);

// Инструменты мониторинга
int activeCount = executor.getActiveCount(); // Сколько потоков "в работе" прямо сейчас
long completedTaskCount = executor.getCompletedTaskCount(); // Сколько задач уже выполнено
int poolSize = executor.getPoolSize(); // Текущий размер пула
int largestPoolSize = executor.getLargestPoolSize(); // Пиковый размер с момента создания
long taskCount = executor.getTaskCount(); // Общее число задач (включая ожидающие)
int queueSize = executor.getQueue().size(); // Размер очереди ожидания

public class AdaptiveExecutor extends ThreadPoolExecutor {
private final ScheduledExecutorService monitor = Executors.newSingleThreadScheduledExecutor();
private final AtomicInteger rejectedCount = new AtomicInteger(0);
 
public AdaptiveExecutor(int corePoolSize, int maximumPoolSize) {
super(corePoolSize, maximumPoolSize, 60, TimeUnit.SECONDS, 
         new LinkedBlockingQueue<>(1000));
         
// Запускаем мониторинг и автоадаптацию каждые 30 секунд
monitor.scheduleAtFixedRate(this::adapt, 30, 30, TimeUnit.SECONDS);
}
 
@Override
protected void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
rejectedCount.incrementAndGet(); // Считаем отклонённые задачи
super.rejectedExecution(r, e);
}
 
private void adapt() {
int rejected = rejectedCount.getAndSet(0);
double utilization = (double) getActiveCount() / getCorePoolSize();
 
// Если много отверженных задач и высокая утилизация - расширяем
if (rejected > 10 && utilization > 0.75 && getCorePoolSize() < getMaximumPoolSize()) {
setCorePoolSize(getCorePoolSize() + 1);
System.out.println("Расширение пула до " + getCorePoolSize());
}
 
// Если мало задач и низкая утилизация - сокращаем
if (rejected == 0 && utilization < 0.25 && getCorePoolSize() > 2) {
setCorePoolSize(getCorePoolSize() - 1);
System.out.println("Сокращение пула до " + getCorePoolSize());
}
}
}

ThreadPoolExecutor executor = new ThreadPoolExecutor(
5, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100),
new ThreadFactory() {
private final AtomicInteger counter = new AtomicInteger(0);
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r, "CustomWorker-" + counter.incrementAndGet());
thread.setDaemon(true); // Поток-демон завершится с завершением программы
thread.setPriority(Thread.NORM_PRIORITY + 1); // Чуть выше обычного
thread.setUncaughtExceptionHandler((t, e) -> {
System.err.println("Необработаное исключение в потоке " + t.getName());
e.printStackTrace();
});
return thread;
}
});

ThreadFactory factory = new ThreadFactoryBuilder()
.setNameFormat("order-processor-%d")
.setDaemon(true)
.setPriority(Thread.MAX_PRIORITY)
.setUncaughtExceptionHandler((t, e) -> logger.error("Error in thread " + t.getName(), e))
.build();

executor.shutdown(); // Мягкое завершение - ждём завершения текущих задач
try {
if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
executor.shutdownNow(); // Жёсткое завершение, если мягкое затянулось
}
} catch (InterruptedException e) {
executor.shutdownNow();
Thread.currentThread().interrupt();
}

ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);
 
// Выполнить задачу через 5 секунд
scheduler.schedule(
    () -> System.out.println("Отложенная задача"),
    5, TimeUnit.SECONDS
);
 
// Выполнять каждые 10 секунд, начиная через 0 секунд
scheduler.scheduleAtFixedRate(
    () -> System.out.println("Повторяющаяся задача - фиксированный интервал"),
    0, 10, TimeUnit.SECONDS
);
 
// Выполнять каждые 10 секунд ПОСЛЕ завершения предыдущего запуска
scheduler.scheduleWithFixedDelay(
    () -> System.out.println("Повторяющаяся задача - фиксированная задержка"),
    0, 10, TimeUnit.SECONDS
);

CompletableFuture<Order> orderFuture = CompletableFuture.supplyAsync(() -> {
    System.out.println("Обрабатываем заказ...");
    return new Order(123, "Книга");
}, executor);
 
// После обработки заказа отправляем email
CompletableFuture<Void> emailFuture = orderFuture.thenAcceptAsync(order -> {
    System.out.println("Отправляем email о заказе " + order.getId());
}, executor);
 
// После email отправляем SMS
CompletableFuture<Void> smsFuture = emailFuture.thenRunAsync(() -> {
    System.out.println("Отправляем SMS");
}, executor);
 
// Параллельно проверяем товар на складе
CompletableFuture<Inventory> inventoryFuture = CompletableFuture.supplyAsync(() -> {
    System.out.println("Проверяем наличие на складе");
    return new Inventory(true);
}, executor);
 
// Объединяем результаты: когда и заказ обработан, и склад проверен
CompletableFuture<Shipment> shipmentFuture = orderFuture
    .thenCombineAsync(inventoryFuture, (order, inventory) -> {
        System.out.println("Создаем отгрузку");
        return new Shipment(order, new Date());
    }, executor);

CompletableFuture<String> future = CompletableFuture
    .supplyAsync(() -> {
        if (Math.random() < 0.5) throw new RuntimeException("Упс!");
        return "Результат";
    })
    .exceptionally(ex -> {
        System.err.println("Произошла ошибка: " + ex.getMessage());
        return "Запасной результат";
    })
    .handle((result, ex) -> {
        if (ex != null) {
            return "Обработано исключение: " + ex.getMessage();
        }
        return "Успешно: " + result;
    });

public class CronScheduler {
private final ScheduledExecutorService scheduler;
private final ConcurrentMap<String, ScheduledFuture<?>> scheduledTasks = new ConcurrentHashMap<>();
 
public CronScheduler(int poolSize) {
scheduler = Executors.newScheduledThreadPool(poolSize);
}
 
public void scheduleTask(String taskId, Runnable task, String cronExpression) {
CronTrigger trigger = new CronTrigger(cronExpression);
Date nextExecutionTime = trigger.nextExecutionTime(new Date());
if (nextExecutionTime == null) {
throw new IllegalArgumentException("Невалидное cron-выражение: " + cronExpression);
}
 
long initialDelay = nextExecutionTime.getTime() - System.currentTimeMillis();
ScheduledFuture<?> future = scheduler.schedule(() -> {
executeTaskAndReschedule(taskId, task, trigger);
}, initialDelay, TimeUnit.MILLISECONDS);
 
scheduledTasks.put(taskId, future);
}
 
private void executeTaskAndReschedule(String taskId, Runnable task, CronTrigger trigger) {
try {
task.run();
} catch (Exception e) {
System.err.println("Ошибка при выполнении задачи " + taskId + ": " + e.getMessage());
} finally {
Date nextExecutionTime = trigger.nextExecutionTime(new Date());
if (nextExecutionTime != null) {
long delay = nextExecutionTime.getTime() - System.currentTimeMillis();
ScheduledFuture<?> future = scheduler.schedule(() -> {
executeTaskAndReschedule(taskId, task, trigger);
}, delay, TimeUnit.MILLISECONDS);
scheduledTasks.put(taskId, future);
}
}
}
 
public void cancelTask(String taskId) {
ScheduledFuture<?> future = scheduledTasks.remove(taskId);
if (future != null) {
future.cancel(false);
}
}
 
// Класс CronTrigger реализует разбор cron-выражений
// и расчет следующего времени запуска
}

public class ExternalSchedulerAdapter {
private final RestTemplate restClient = new RestTemplate();
private final String schedulerEndpoint;
private final ExecutorService executor;
 
public ExternalSchedulerAdapter(String schedulerEndpoint, ExecutorService executor) {
this.schedulerEndpoint = schedulerEndpoint;
this.executor = executor;
}
 
public String scheduleTask(TaskRequest request) {
return restClient.postForObject(schedulerEndpoint + "/schedule", request, ScheduleResponse.class).getTaskId();
}
 
public void registerCallback(String taskId, Runnable callback) {
TaskRegistry.getInstance().registerTask(taskId, callback);
}
 
// Веб-контроллер для принятия колбэков от внешнего планировщика
@RestController
public static class CallbackController {
@PostMapping("/tasks/{taskId}/execute")
public ResponseEntity<String> executeTask(@PathVariable String taskId) {
Runnable task = TaskRegistry.getInstance().getTask(taskId);
if (task == null) {
return ResponseEntity.notFound().build();
}
    
CompletableFuture.runAsync(task, executor);
return ResponseEntity.ok("Task execution initiated");
}
}
 
// Синглтон-реестр задач
private static class TaskRegistry {
private static final TaskRegistry INSTANCE = new TaskRegistry();
private final ConcurrentMap<String, Runnable> tasks = new ConcurrentHashMap<>();
    
public static TaskRegistry getInstance() {
return INSTANCE;
}
    
public void registerTask(String id, Runnable task) {
tasks.put(id, task);
}
    
public Runnable getTask(String id) {
return tasks.get(id);
}
}
}

public class RetryableScheduledTask implements Runnable {
    private final Runnable task;
    private final int maxRetries;
    private final long retryDelayMs;
    private final ScheduledExecutorService scheduler;
    private int retryCount = 0;
 
    public RetryableScheduledTask(Runnable task, int maxRetries, 
                                 long retryDelayMs, 
                                 ScheduledExecutorService scheduler) {
        this.task = task;
        this.maxRetries = maxRetries;
        this.retryDelayMs = retryDelayMs;
        this.scheduler = scheduler;
    }
 
    @Override
    public void run() {
        try {
            task.run();
            // Сброс счетчика при успехе
            retryCount = 0;
        } catch (Exception e) {
            retryCount++;
            if (retryCount <= maxRetries) {
                System.err.println("Задача не выполнена. Повторная попытка " 
                                  + retryCount + "/" + maxRetries);
                scheduler.schedule(this, retryDelayMs, TimeUnit.MILLISECONDS);
            } else {
                System.err.println("Задача окончательно не выполнена после " 
                                  + maxRetries + " попыток");
                // Здесь можно добавить доп. обработку
            }
        }
    }
}

public class MonitoredScheduledExecutor extends ScheduledThreadPoolExecutor {
    private final Counter taskStartedCounter;
    private final Counter taskCompletedCounter;
    private final Counter taskFailedCounter;
    private final Timer taskDurationTimer;
    
    public MonitoredScheduledExecutor(int corePoolSize, MetricRegistry metrics) {
        super(corePoolSize);
        this.taskStartedCounter = metrics.counter("tasks.started");
        this.taskCompletedCounter = metrics.counter("tasks.completed");
        this.taskFailedCounter = metrics.counter("tasks.failed");
        this.taskDurationTimer = metrics.timer("tasks.duration");
    }
    
    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        super.beforeExecute(t, r);
        taskStartedCounter.inc();
    }
    
    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        try {
            long startTime = ((MonitoredTask)r).getStartTime();
            long duration = System.currentTimeMillis() - startTime;
            taskDurationTimer.update(duration, TimeUnit.MILLISECONDS);
            
            if (t == null) {
                taskCompletedCounter.inc();
            } else {
                taskFailedCounter.inc();
            }
        } finally {
            super.afterExecute(r, t);
        }
    }
}

public class ExpiringCache<K, V> {
    private final ConcurrentHashMap<K, CacheEntry<V>> cache = new ConcurrentHashMap<>();
    private final ScheduledExecutorService cleanupService;
    
    public ExpiringCache() {
        cleanupService = Executors.newSingleThreadScheduledExecutor(r -> {
            Thread t = new Thread(r, "cache-cleanup-thread");
            t.setDaemon(true);
            return t;
        });
        
        // Запускаем периодическую очистку
        cleanupService.scheduleAtFixedRate(this::cleanup, 5, 5, TimeUnit.MINUTES);
    }
    
    public void put(K key, V value, long expirationMs) {
        cache.put(key, new CacheEntry<>(value, System.currentTimeMillis() + expirationMs));
    }
    
    public V get(K key) {
        CacheEntry<V> entry = cache.get(key);
        if (entry == null) return null;
        
        // Проверяем истечение срока при каждом доступе
        if (entry.isExpired()) {
            cache.remove(key);
            return null;
        }
        
        return entry.getValue();
    }
    
    private void cleanup() {
        long now = System.currentTimeMillis();
        cache.entrySet().removeIf(entry -> entry.getValue().getExpiryTime() <= now);
    }
    
    private static class CacheEntry<V> {
        private final V value;
        private final long expiryTime;
        
        public CacheEntry(V value, long expiryTime) {
            this.value = value;
            this.expiryTime = expiryTime;
        }
        
        public V getValue() {
            return value;
        }
        
        public long getExpiryTime() {
            return expiryTime;
        }
        
        public boolean isExpired() {
            return System.currentTimeMillis() > expiryTime;
        }
    }
}

public class ProducerConsumerExample {
    public static void main(String[] args) {
        // Создание буферизованой очереди ограниченного размера
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(10);
        
        // Создаем кастомный пул для потребителей
        ThreadPoolExecutor consumerPool = new ThreadPoolExecutor(
            3, 5, 60, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(100),
            new ThreadFactoryBuilder().setNameFormat("consumer-%d").build(),
            new ThreadPoolExecutor.CallerRunsPolicy()
        );
        
        // Запускаем нескольких потребителей
        for (int i = 0; i < 3; i++) {
            consumerPool.submit(() -> {
                try {
                    while (!Thread.currentThread().isInterrupted()) {
                        String item = queue.take(); // Блокируемся, пока нет элементов
                        process(item);
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }
        
        // Производители данных
        ScheduledExecutorService producerService = Executors.newSingleThreadScheduledExecutor();
        producerService.scheduleAtFixedRate(() -> {
            try {
                String data = generateData();
                queue.put(data); // Блокируемся, если очередь полна
                System.out.println("Добавлено: " + data + " [Размер очереди: " + queue.size() + "]");
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }, 0, 100, TimeUnit.MILLISECONDS);
    }
    
    private static String generateData() {
        return "Item-" + System.nanoTime();
    }
    
    private static void process(String item) {
        try {
            // Имитация долгой обработки
            Thread.sleep(300);
            System.out.println(Thread.currentThread().getName() + " обработан: " + item);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

public class ParallelDataProcessor {
    private final ForkJoinPool forkJoinPool;
    
    public ParallelDataProcessor(int parallelism) {
        this.forkJoinPool = new ForkJoinPool(
            parallelism,
            ForkJoinPool.defaultForkJoinWorkerThreadFactory,
            (t, e) -> System.err.println("Uncaught error: " + e),
            true // Асинхронный режим
        );
    }
    
    public <T, R> List<R> processDataParallel(List<T> items, Function<T, R> processor) {
        return forkJoinPool.submit(() ->
            items.parallelStream()
                .map(processor)
                .collect(Collectors.toList())
        ).join();
    }
    
    public <T, R> List<R> processDataParallelWithThreshold(List<T> items, 
                                                        Function<T, R> processor,
                                                        int threshold) {
        return forkJoinPool.invoke(new ProcessingTask<>(items, processor, threshold));
    }
    
    private static class ProcessingTask<T, R> extends RecursiveTask<List<R>> {
        private final List<T> items;
        private final Function<T, R> processor;
        private final int threshold;
        
        public ProcessingTask(List<T> items, Function<T, R> processor, int threshold) {
            this.items = items;
            this.processor = processor;
            this.threshold = threshold;
        }
        
        @Override
        protected List<R> compute() {
            if (items.size() <= threshold) {
                return items.stream()
                    .map(processor)
                    .collect(Collectors.toList());
            }
            
            int mid = items.size() / 2;
            ProcessingTask<T, R> leftTask = new ProcessingTask<>(
                items.subList(0, mid), processor, threshold);
            ProcessingTask<T, R> rightTask = new ProcessingTask<>(
                items.subList(mid, items.size()), processor, threshold);
                
            // Запускаем левую задачу асинхронно
            leftTask.fork();
            
            // Обрабатываем правую часть в текущем потоке
            List<R> rightResult = rightTask.compute();
            
            // Ждем результата левой задачи
            List<R> leftResult = leftTask.join();
            
            // Объединяем результаты
            List<R> result = new ArrayList<>(leftResult);
            result.addAll(rightResult);
            return result;
        }
    }
}

public class RequestContextExample {
    // ThreadLocal переменная хранит данные отдельно для каждого потока
    private static final ThreadLocal<RequestContext> requestContext = ThreadLocal.withInitial(RequestContext::new);
    
    // Зааменяем дорогой DateFormatter на ThreadLocal вариант для лучшей производительности
    private static final ThreadLocal<DateTimeFormatter> formatter = 
        ThreadLocal.withInitial(() -> DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
    
    public static RequestContext getContext() {
        return requestContext.get();
    }
    
    public static void clearContext() {
        requestContext.remove(); // Важно предотвратить утечки памяти
    }
    
    public static String formatDate(LocalDateTime date) {
        return formatter.get().format(date);
    }
    
    // В реальном веб-приложении этот метод вызывается перед обработкой запроса
    public static void processRequest(String userId, HttpRequest request) {
        try {
            RequestContext context = getContext();
            context.setUserId(userId);
            context.setRequestTime(LocalDateTime.now());
            
            // Обработка запроса с доступом к контексту
            handleRequest(request);
        } finally {
            // Очистка после каждого запроса
            clearContext();
        }
    }
    
    private static void handleRequest(HttpRequest request) {
        // Доступ к контексту из любого места в процессе обработки
        RequestContext context = getContext();
        System.out.println("Обработка запроса для пользователя: " + context.getUserId() + 
                           " в " + formatDate(context.getRequestTime()));
    }
    
    public static class RequestContext {
        private String userId;
        private LocalDateTime requestTime;
        
        // Геттеры и сеттеры
        public String getUserId() { return userId; }
        public void setUserId(String userId) { this.userId = userId; }
        public LocalDateTime getRequestTime() { return requestTime; }
        public void setRequestTime(LocalDateTime requestTime) { this.requestTime = requestTime; }
    }
}

public class CircuitBreaker {
private final int failureThreshold;
private final long resetTimeoutMs;
private final AtomicInteger failureCount = new AtomicInteger(0);
private final AtomicReference<CircuitState> state = new AtomicReference<>(CircuitState.CLOSED);
private final AtomicLong lastFailureTime = new AtomicLong(0);
 
public CircuitBreaker(int failureThreshold, long resetTimeoutMs) {
    this.failureThreshold = failureThreshold;
    this.resetTimeoutMs = resetTimeoutMs;
}
 
public <T> T execute(Supplier<T> operation) throws Exception {
    if (state.get() == CircuitState.OPEN) {
        // Проверяем, прошло ли достаточно времени для перехода в полуоткрытое состояние
        if (System.currentTimeMillis() - lastFailureTime.get() > resetTimeoutMs) {
            synchronized (this) {
                if (state.get() == CircuitState.OPEN) {
                    state.set(CircuitState.HALF_OPEN);
                }
            }
        } else {
            throw new CircuitBreakerOpenException("Предохранитель разомкнут");
        }
    }
    
    try {
        T result = operation.get();
        reset(); // Успешное выполнение - сбрасываем счетчик
        return result;
    } catch (Exception e) {
        recordFailure();
        throw e;
    }
}
 
private void recordFailure() {
    lastFailureTime.set(System.currentTimeMillis());
    if (state.get() == CircuitState.HALF_OPEN) {
        trip();
        return;
    }
    
    if (failureCount.incrementAndGet() >= failureThreshold) {
        trip();
    }
}
 
private void trip() {
    state.set(CircuitState.OPEN);
}
 
private void reset() {
    failureCount.set(0);
    state.set(CircuitState.CLOSED);
}
 
private enum CircuitState {
    CLOSED,     // Нормальная работа
    OPEN,       // Предохранитель сработал, операции блокируются
    HALF_OPEN   // Тестовый режим для проверки восстановления
}
 
public static class CircuitBreakerOpenException extends RuntimeException {
    public CircuitBreakerOpenException(String message) {
        super(message);
    }
}
}

public class BulkheadExample {
// Отдельный пул для критически важных операций
private final ExecutorService criticalPool = new ThreadPoolExecutor(
    5, 10, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100),
    new ThreadFactoryBuilder().setNameFormat("critical-ops-%d").build()
);
 
// Пул для некритичных фоновых операций
private final ExecutorService backgroundPool = new ThreadPoolExecutor(
    2, 5, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(500),
    new ThreadFactoryBuilder().setNameFormat("background-ops-%d").build()
);
 
public CompletableFuture<OrderStatus> processOrder(Order order) {
    return CompletableFuture.supplyAsync(() -> {
        // Критически важный бизнес-процесс
        return calculateOrderStatus(order);
    }, criticalPool);
}
 
public void sendNotification(User user, String message) {
    CompletableFuture.runAsync(() -> {
        // Некритичная фоновая операция
        notificationService.send(user, message);
    }, backgroundPool);
}
}

// Антипаттерн: блокировка потока на длительные IO-операции
CompletableFuture.supplyAsync(() -> {
    try {
        // Блокируем поток на время HTTP-запроса
        return httpClient.send(request, HttpResponse.BodyHandlers.ofString()).body();
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
}, executor);
 
// Правильный подход: асинхронные неблокирующие операции
httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofString())
    .thenApplyAsync(HttpResponse::body, executor);

// Старый подход: пул потоков ограничен размером
ExecutorService executorService = Executors.newFixedThreadPool(100);
 
// Новый подход: виртуальные потоки практически бесконечны
ExecutorService virtualExecutor = Executors.newVirtualThreadPerTaskExecutor();
 
virtualExecutor.submit(() -> {
    // Выполняем блокирующую операцию без угрызений совести
    Thread.sleep(1000);
    return "Готово";
});