Зачем нужны потоки
Поток — независимая последовательность выполнения внутри процесса. Несколько потоков позволяют:
- использовать все ядра процессора;
- не блокировать интерфейс или сервер во время долгих операций;
- обрабатывать много запросов одновременно.
Потоки разделяют общую память. Отсюда главная сложность: одновременный доступ к одним данным.
Создание потока
public class ThreadBasics { public static void main(String[] args) throws InterruptedException { Runnable task = () -> { String name = Thread.currentThread().getName(); for (int i = 1; i <= 3; i++) { System.out.println(name + ": шаг " + i); } }; Thread a = new Thread(task, "Поток-A"); Thread b = new Thread(task, "Поток-B"); a.start(); b.start(); a.join(); b.join(); System.out.println("Оба потока завершены"); }}start()запускает новый поток. Вызовrun()напрямую выполнит код в текущем потоке.join()ждет завершения потока.- Порядок вывода потоков при каждом запуске может отличаться.
Наследование от Thread:
public class Worker extends Thread { @Override public void run() { System.out.println("Работает " + getName()); } public static void main(String[] args) { new Worker().start(); }}Предпочтительнее передавать Runnable: так задача отделена от механизма запуска.
Sleep и прерывание
public class Interrupts { public static void main(String[] args) throws InterruptedException { Thread worker = new Thread(() -> { try { while (true) { System.out.println("работаю"); Thread.sleep(500); } } catch (InterruptedException e) { System.out.println("прерван"); } }); worker.start(); Thread.sleep(1600); worker.interrupt(); worker.join(); }}interrupt() — запрос на остановку. Поток должен сам обработать InterruptedException или проверять Thread.currentThread().isInterrupted(). Не проглатывайте исключение молча: либо завершайте работу, либо восстанавливайте флаг Thread.currentThread().interrupt().
Состояние гонки
public class RaceCondition { static int counter = 0; public static void main(String[] args) throws InterruptedException { Runnable task = () -> { for (int i = 0; i < 100_000; i++) { counter++; } }; Thread a = new Thread(task); Thread b = new Thread(task); a.start(); b.start(); a.join(); b.join(); System.out.println(counter); }}Ожидается 200000, но результат почти всегда меньше. Операция counter++ состоит из трех шагов: чтение, увеличение, запись. Потоки перемешивают эти шаги и теряют обновления.
synchronized
Блок или метод synchronized может выполнять только один поток одновременно для одного монитора:
public class SafeCounter { private int counter = 0; public synchronized void increment() { counter++; } public synchronized int get() { return counter; } public static void main(String[] args) throws InterruptedException { SafeCounter safe = new SafeCounter(); Runnable task = () -> { for (int i = 0; i < 100_000; i++) { safe.increment(); } }; Thread a = new Thread(task); Thread b = new Thread(task); a.start(); b.start(); a.join(); b.join(); System.out.println(safe.get()); }}Теперь результат всегда 200000.
Синхронизация блока с явным объектом-монитором:
public class Account { private final Object lock = new Object(); private double balance; public void deposit(double amount) { synchronized (lock) { balance += amount; } } public double getBalance() { synchronized (lock) { return balance; } }}Синхронизация одновременно решает две задачи: взаимное исключение и видимость изменений между потоками.
volatile
Модификатор volatile гарантирует видимость записи другими потоками, но не атомарность составных операций:
public class StopFlag { private volatile boolean running = true; public void stop() { running = false; } public void runLoop() { while (running) { Thread.onSpinWait(); } System.out.println("Остановлено"); } public static void main(String[] args) throws InterruptedException { StopFlag flag = new StopFlag(); Thread t = new Thread(flag::runLoop); t.start(); Thread.sleep(200); flag.stop(); t.join(); }}Без volatile поток мог бы никогда не увидеть изменение флага.
Атомарные типы
Для простых счетчиков не нужна блокировка:
import java.util.concurrent.atomic.AtomicInteger;import java.util.concurrent.atomic.AtomicLong;public class AtomicDemo { public static void main(String[] args) throws InterruptedException { AtomicInteger counter = new AtomicInteger(); Runnable task = () -> { for (int i = 0; i < 100_000; i++) { counter.incrementAndGet(); } }; Thread a = new Thread(task); Thread b = new Thread(task); a.start(); b.start(); a.join(); b.join(); System.out.println(counter.get()); AtomicLong value = new AtomicLong(5); System.out.println(value.compareAndSet(5, 10)); System.out.println(value.get()); System.out.println(value.updateAndGet(v -> v * 2)); }}Блокировки Lock
import java.util.concurrent.locks.Lock;import java.util.concurrent.locks.ReentrantLock;public class LockDemo { private final Lock lock = new ReentrantLock(); private int value; public void increment() { lock.lock(); try { value++; } finally { lock.unlock(); } } public boolean tryIncrement() { if (lock.tryLock()) { try { value++; return true; } finally { lock.unlock(); } } return false; } public int get() { lock.lock(); try { return value; } finally { lock.unlock(); } }}unlock() обязательно вызывать в finally.
Взаимная блокировка
Deadlock возникает, когда потоки ждут друг друга бесконечно:
public class Deadlock { static final Object A = new Object(); static final Object B = new Object(); public static void main(String[] args) { new Thread(() -> { synchronized (A) { sleep(); synchronized (B) { System.out.println("поток 1"); } } }).start(); new Thread(() -> { synchronized (B) { sleep(); synchronized (A) { System.out.println("поток 2"); } } }).start(); } static void sleep() { try { Thread.sleep(100); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }}Программа зависнет. Как избежать: всегда захватывать блокировки в одном и том же порядке, использовать tryLock с таймаутом, держать блокировки как можно короче. Для диагностики зависшей программы используйте jstack <pid>.
Пул потоков и ExecutorService
Создавать поток на каждую задачу дорого. Пул переиспользует потоки:
import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.TimeUnit;public class ExecutorDemo { public static void main(String[] args) throws InterruptedException { ExecutorService pool = Executors.newFixedThreadPool(3); for (int i = 1; i <= 6; i++) { int id = i; pool.submit(() -> { System.out.println("Задача " + id + " в " + Thread.currentThread().getName()); try { Thread.sleep(200); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }); } pool.shutdown(); if (!pool.awaitTermination(5, TimeUnit.SECONDS)) { pool.shutdownNow(); } }}Пул обязательно нужно останавливать (shutdown), иначе программа не завершится.
Типы пулов:
| Фабричный метод | Назначение |
|---|---|
newFixedThreadPool(n) | фиксированное число потоков |
newCachedThreadPool() | создает потоки по необходимости |
newSingleThreadExecutor() | один поток, задачи по очереди |
newScheduledThreadPool(n) | задачи по расписанию |
newVirtualThreadPerTaskExecutor() | виртуальный поток на задачу (Java 21) |
С Java 19 ExecutorService реализует AutoCloseable, поэтому можно писать так:
import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;public class ExecutorTry { public static void main(String[] args) { try (ExecutorService pool = Executors.newFixedThreadPool(2)) { pool.submit(() -> System.out.println("задача 1")); pool.submit(() -> System.out.println("задача 2")); } }}Выход из блока try ждет завершения задач.
Callable и Future
Runnable ничего не возвращает. Callable<T> возвращает результат и может бросать исключения:
import java.util.ArrayList;import java.util.List;import java.util.concurrent.Callable;import java.util.concurrent.ExecutionException;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.Future;import java.util.concurrent.TimeUnit;import java.util.concurrent.TimeoutException;public class FutureDemo { static long sumRange(int from, int to) { long sum = 0; for (int i = from; i <= to; i++) { sum += i; } return sum; } public static void main(String[] args) throws InterruptedException, ExecutionException { ExecutorService pool = Executors.newFixedThreadPool(4); List<Future<Long>> futures = new ArrayList<>(); for (int part = 0; part < 4; part++) { int from = part * 250_000 + 1; int to = (part + 1) * 250_000; Callable<Long> task = () -> sumRange(from, to); futures.add(pool.submit(task)); } long total = 0; for (Future<Long> future : futures) { total += future.get(); } System.out.println(total); Future<String> slow = pool.submit(() -> { Thread.sleep(2000); return "готово"; }); try { System.out.println(slow.get(500, TimeUnit.MILLISECONDS)); } catch (TimeoutException e) { System.out.println("Таймаут"); slow.cancel(true); } pool.shutdown(); }}Метод get() блокирует вызывающий поток до получения результата. Исключение из задачи приходит как ExecutionException.
CompletableFuture
Позволяет строить асинхронные цепочки без блокировки:
import java.util.concurrent.CompletableFuture;import java.util.concurrent.ExecutionException;public class CompletableDemo { static String loadUser(int id) { sleep(300); return "Пользователь-" + id; } static int loadScore(String user) { sleep(300); return user.length() * 10; } static void sleep(long ms) { try { Thread.sleep(ms); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } public static void main(String[] args) throws ExecutionException, InterruptedException { CompletableFuture<Integer> pipeline = CompletableFuture .supplyAsync(() -> loadUser(7)) .thenApply(user -> user.toUpperCase()) .thenApply(CompletableDemo::loadScore); System.out.println(pipeline.get()); CompletableFuture<String> a = CompletableFuture.supplyAsync(() -> loadUser(1)); CompletableFuture<String> b = CompletableFuture.supplyAsync(() -> loadUser(2)); CompletableFuture<String> both = a.thenCombine(b, (x, y) -> x + " и " + y); System.out.println(both.join()); CompletableFuture<Integer> failing = CompletableFuture .supplyAsync(() -> { if (true) { throw new IllegalStateException("сбой"); } return 1; }) .exceptionally(e -> -1); System.out.println(failing.join()); CompletableFuture<Void> all = CompletableFuture.allOf(a, b); all.join(); System.out.println("Все завершены"); }}Методы:
supplyAsync,runAsync— запуск асинхронной задачи;thenApply— преобразовать результат;thenAccept— потребить результат;thenCompose— продолжить другой асинхронной задачей;thenCombine— объединить два результата;exceptionally,handle— обработка ошибок;allOf,anyOf— ожидание нескольких.
По умолчанию задачи выполняются в общем пуле ForkJoinPool.commonPool(). Для блокирующих операций передавайте собственный Executor вторым аргументом.
Потокобезопасные коллекции
Обычные ArrayList и HashMap не потокобезопасны. Используйте:
import java.util.List;import java.util.Map;import java.util.concurrent.ConcurrentHashMap;import java.util.concurrent.CopyOnWriteArrayList;public class ConcurrentCollections { public static void main(String[] args) throws InterruptedException { Map<String, Integer> counts = new ConcurrentHashMap<>(); List<String> log = new CopyOnWriteArrayList<>(); Runnable task = () -> { for (int i = 0; i < 1000; i++) { counts.merge("key", 1, Integer::sum); } log.add(Thread.currentThread().getName()); }; Thread a = new Thread(task); Thread b = new Thread(task); a.start(); b.start(); a.join(); b.join(); System.out.println(counts.get("key")); System.out.println(log.size()); }}Очередь для передачи данных между потоками:
import java.util.concurrent.ArrayBlockingQueue;import java.util.concurrent.BlockingQueue;public class ProducerConsumer { public static void main(String[] args) throws InterruptedException { BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(5); Thread producer = new Thread(() -> { try { for (int i = 1; i <= 10; i++) { queue.put(i); System.out.println("Произведено " + i); } queue.put(-1); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }); Thread consumer = new Thread(() -> { try { while (true) { int value = queue.take(); if (value == -1) { break; } System.out.println("Потреблено " + value); Thread.sleep(50); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }); producer.start(); consumer.start(); producer.join(); consumer.join(); }}put блокируется при заполненной очереди, take — при пустой.
Синхронизаторы
import java.util.concurrent.CountDownLatch;public class LatchDemo { public static void main(String[] args) throws InterruptedException { int workers = 3; CountDownLatch latch = new CountDownLatch(workers); for (int i = 1; i <= workers; i++) { int id = i; new Thread(() -> { System.out.println("Работник " + id + " закончил"); latch.countDown(); }).start(); } latch.await(); System.out.println("Все работники закончили"); }}Другие классы: Semaphore (ограничение числа одновременных доступов), CyclicBarrier (ожидание всех участников в точке), Phaser.
Виртуальные потоки
В Java 21 появились виртуальные потоки: очень легкие потоки, которыми управляет JVM. Они удобны для большого числа блокирующих задач (сеть, база):
import java.time.Duration;import java.util.concurrent.Executors;import java.util.stream.IntStream;public class VirtualThreads { public static void main(String[] args) { try (var executor = Executors.newVirtualThreadPerTaskExecutor()) { IntStream.range(0, 10_000).forEach(i -> executor.submit(() -> { Thread.sleep(Duration.ofSeconds(1)); return i; })); } System.out.println("Готово"); }}Одиночный виртуальный поток:
Thread.startVirtualThread(() -> System.out.println("виртуальный поток"));10 000 обычных потоков уничтожили бы память, а виртуальные справляются без проблем. Пулировать виртуальные потоки не нужно.
Практические правила
- Если можно, избегайте общего изменяемого состояния: используйте неизменяемые объекты.
- Предпочитайте высокоуровневые средства (
ExecutorService,CompletableFuture,ConcurrentHashMap) ручной работе сThreadиsynchronized. - Всегда останавливайте пулы потоков.
- Не игнорируйте
InterruptedException. - Держите критические секции короткими, не вызывайте внешний код внутри
synchronized. - Тестируйте многопоточный код многократно: ошибки гонки проявляются не всегда.
- Параллельность не всегда ускоряет программу: накладные расходы могут перевесить выгоду.