Как потоки взаимодействуют между собой

# Взаимодействие потоков (Threads) в Java

Основные механизмы

Потоки взаимодействуют через:

1. Shared Memory (общая память)
2. Synchronization (синхронизация)
3. Communication (обмен сообщениями)

1. Shared Memory (Race Condition)

Проблема: Race Condition

public class Counter {
    private int count = 0;  // Общая память для всех потоков
    
    public void increment() {
        count++;  // ❌ НЕ АТОМАРНАЯ операция!
    }
}

// Использование
Counter counter = new Counter();

// Поток 1
for (int i = 0; i < 1000; i++) {
    counter.increment();
}

// Поток 2
for (int i = 0; i < 1000; i++) {
    counter.increment();
}

// Результат: ~1234 вместо 2000
// Потому что count++ на самом деле:
// 1. READ: int temp = count;     (могут прерватьсь отдельно)
// 2. ADD:  temp = temp + 1;
// 3. WRITE: count = temp;

2. Synchronized (Мьютекс/Монитор)

Synchronized метод

public class Counter {
    private int count = 0;
    
    public synchronized void increment() {  // Заблокирован для остальных потоков
        count++;
    }
    
    public synchronized int getCount() {
        return count;
    }
}

// Теперь результат = 2000 (правильный)
// Но потоки выстраиваются в очередь

Synchronized блок (более гибкий)

public class Account {
    private int balance = 0;
    private Object lock = new Object();
    
    public void transfer(int amount) {
        synchronized(lock) {  // Блокируем только критическую секцию
            balance += amount;
        }
        // Остальной код может выполняться параллельно
    }
}

3. Volatile (видимость памяти)

public class MyThread implements Runnable {
    private volatile boolean running = true;  // Видимо для всех потоков
    
    public void run() {
        while (running) {
            System.out.println("Running");
        }
    }
    
    public void stop() {
        running = false;  // Другой поток сразу это увидит
    }
}

MyThread thread = new MyThread();
new Thread(thread).start();
Thread.sleep(1000);
thread.stop();  // Поток остановится

volatile vs synchronized

• volatile: только видимость, НЕ атомарность
• synchronized: видимость + атомарность, но медленнее

private volatile int count = 0;
count++;  // ❌ Всё ещё race condition!

// Нужно:
private int count = 0;
synchronized void increment() {
    count++;  // ✅ Безопасно
}

4. Atomic операции (Lock-free)

public class Counter {
    private AtomicInteger count = new AtomicInteger(0);
    
    public void increment() {
        count.incrementAndGet();  // Атомарная операция
    }
    
    public int getCount() {
        return count.get();
    }
}

// Быстрее чем synchronized (используют CAS — Compare-And-Swap)
// Результат = 2000 (правильный)

5. wait() / notify() (условные переменные)

public class Queue<T> {
    private List<T> items = new ArrayList<>();
    private int maxSize = 10;
    
    // Производитель
    public synchronized void put(T item) throws InterruptedException {
        while (items.size() >= maxSize) {
            wait();  // Ждёт пока потребитель что-то возьмёт
        }
        items.add(item);
        notifyAll();  // Сообщает потребителю: есть данные!
    }
    
    // Потребитель
    public synchronized T take() throws InterruptedException {
        while (items.isEmpty()) {
            wait();  // Ждёт пока производитель что-то добавит
        }
        T item = items.remove(0);
        notifyAll();  // Сообщает производителю: место в очереди!
        return item;
    }
}

6. BlockingQueue (очередь между потоками)

// Потокобезопасная очередь для взаимодействия
BlockingQueue<Integer> queue = new LinkedBlockingQueue<>();

// Поток-производитель
new Thread(() -> {
    try {
        for (int i = 0; i < 100; i++) {
            queue.put(i);  // Добавляет в очередь
            System.out.println("Produced: " + i);
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}).start();

// Поток-потребитель
new Thread(() -> {
    try {
        while (true) {
            Integer value = queue.take();  // Берёт из очереди, ждёт если пусто
            System.out.println("Consumed: " + value);
        }
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}).start();

7. CountDownLatch (ожидание нескольких потоков)

public void downloadFiles() throws InterruptedException {
    int fileCount = 5;
    CountDownLatch latch = new CountDownLatch(fileCount);
    
    // Запускаем потоки-загрузчики
    for (int i = 0; i < fileCount; i++) {
        new Thread(() -> {
            try {
                downloadFile();
            } finally {
                latch.countDown();  // Отчитывается о завершении
            }
        }).start();
    }
    
    latch.await();  // Ждём пока все 5 потоков завершат работу
    System.out.println("All files downloaded");
}

8. Barrier (синхронизация потоков)

public void runPhases() throws Exception {
    CyclicBarrier barrier = new CyclicBarrier(3);  // 3 потока
    
    for (int i = 0; i < 3; i++) {
        new Thread(() -> {
            try {
                System.out.println("Phase 1: " + Thread.currentThread().getName());
                barrier.await();  // Все ждут друг друга
                
                System.out.println("Phase 2: " + Thread.currentThread().getName());
                barrier.await();  // Опять ждут
                
                System.out.println("Phase 3: " + Thread.currentThread().getName());
            } catch (Exception e) {}
        }).start();
    }
}

// Output:
// Phase 1: Thread-0
// Phase 1: Thread-1
// Phase 1: Thread-2
// Phase 2: Thread-0  (только после Phase 1 у всех)
// Phase 2: Thread-1
// Phase 2: Thread-2
// ...

9. Semaphore (ограничение доступа)

public class DatabaseConnection {
    private Semaphore semaphore = new Semaphore(3);  // Только 3 одновременно
    
    public Connection getConnection() throws InterruptedException {
        semaphore.acquire();  // Занимает один "слот"
        return new Connection();
    }
    
    public void releaseConnection(Connection conn) {
        conn.close();
        semaphore.release();  // Освобождает слот
    }
}

// 10 потоков хотят подключиться, но только 3 одновременно

10. Реальный пример: Producer-Consumer Pattern

public class ProducerConsumer {
    private BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(5);
    
    public void start() {
        // Producer
        ExecutorService executor = Executors.newFixedThreadPool(2);
        
        executor.submit(() -> {
            for (int i = 0; i < 20; i++) {
                try {
                    queue.put(i);
                    System.out.println("Produced: " + i);
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        });
        
        // Consumer
        executor.submit(() -> {
            while (true) {
                try {
                    Integer value = queue.take();
                    System.out.println("Consumed: " + value);
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        });
    }
}

Сравнение механизмов

Best Practices

1. Минимизируй общую память — держи потоки отдельными
2. Используй потокобезопасные структуры: ConcurrentHashMap, BlockingQueue
3. Избегай deadlock'ов — всегда блокируй ресурсы в одном порядке
4. Тестируй многопоточный код — используй stress tests
5. Предпочитай high-level утилиты: ExecutorService, Fork/Join вместо raw Thread
6. Документируй thread-safety — укажи какие методы потокобезопасны