Многопоточность в Java: потоки и синхронизация

1// Способ 1: наследование Thread
2class MyThread extends Thread {
3    @Override
4    public void run() {
5        System.out.println("Поток: " + Thread.currentThread().getName());
6    }
7}
8
9// Способ 2: реализация Runnable
10Runnable task = () -> {
11    System.out.println("Задача выполняется");
12};
13new Thread(task).start();
14
15// Способ 3: Callable с возвращаемым значением
16Callable<Integer> callable = () -> {
17    return 42;
18};
19ExecutorService executor = Executors.newSingleThreadExecutor();
20Future<Integer> future = executor.submit(callable);

1// Проблема: гонка данных
2private int counter = 0;
3
4public void increment() {
5    counter++; // Не атомарная операция!
6    // 1. Читает counter
7    // 2. Прибавляет 1
8    // 3. Записывает результат
9}

1// Поток 1:
2synchronized (lockA) {
3    synchronized (lockB) { /* ... */ }
4}
5
6// Поток 2:
7synchronized (lockB) {
8    synchronized (lockA) { /* ... */ } // Deadlock!
9}

1public class Counter {
2    private int count = 0;
3
4    public synchronized void increment() {
5        count++;
6    }
7
8    public synchronized int getCount() {
9        return count;
10    }
11}
12
13// Блокировка на уровне объекта
14public void updateSharedData() {
15    synchronized (this) {
16        // критическая секция
17    }
18}

1private final ReentrantLock lock = new ReentrantLock();
2private final Condition notEmpty = lock.newCondition();
3
4public void produce() {
5    lock.lock();
6    try {
7        // производство данных
8        notEmpty.signal();
9    } finally {
10        lock.unlock();
11    }
12}
13
14public void consume() throws InterruptedException {
15    lock.lock();
16    try {
17        while (isEmpty()) {
18            notEmpty.await();
19        }
20        // потребление данных
21    } finally {
22        lock.unlock();
23    }
24}

1// Фиксированный пул
2ExecutorService fixedPool = Executors.newFixedThreadPool(4);
3
4// Пул с кэшированием
5ExecutorService cachedPool = Executors.newCachedThreadPool();
6
7// Пул сcheduled задач
8ScheduledExecutorService scheduledPool = Executors.newScheduledThreadPool(2);
9
10// Отправка задач
11fixedPool.submit(() -> {
12    System.out.println("Выполняется в пуле потоков");
13});
14
15// Shutdown
16fixedPool.shutdown();
17fixedPool.awaitTermination(60, TimeUnit.SECONDS);

1CompletableFuture.supplyAsync(() -> {
2    return fetchDataFromService();
3})
4.thenApply(data -> {
5    return transformData(data);
6})
7.thenAccept(result -> {
8    saveResult(result);
9})
10.exceptionally(ex -> {
11    log.error("Ошибка: {}", ex.getMessage());
12    return null;
13});

1CompletableFuture<User> userFuture = getUserAsync();
2CompletableFuture<List<Order>> ordersFuture = getOrdersAsync();
3
4// Ждём оба результата
5CompletableFuture<Void> combined = CompletableFuture.allOf(userFuture, ordersFuture);
6combined.join();
7
8User user = userFuture.join();
9List<Order> orders = ordersFuture.join();

1private AtomicInteger atomicCounter = new AtomicInteger(0);
2
3public void increment() {
4    atomicCounter.incrementAndGet();
5}
6
7public int get() {
8    return atomicCounter.get();
9}

1ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();
2
3// Атомарное обновление
4map.compute("key", (k, v) -> v == null ? 1 : v + 1);
5
6// Атомарное добавление если отсутствует
7map.putIfAbsent("key", 1);

1private volatile boolean running = true;
2
3public void stop() {
4    running = false; // Видимо другим потокам
5}
6
7public void run() {
8    while (running) {
9        // работа
10    }
11}

1// parallelStream для параллельной обработки коллекций
2List<Result> results = data.parallelStream()
3    .filter(item -> item.isActive())
4    .map(this::processItem)
5    .collect(Collectors.toList());