Практические паттерны и оптимизация в Go vs Java | Concurrency часть 3

В этой части мы рассмотрим практические паттерны параллельной обработки задач: worker pool, pipeline pattern и схемы сборки результатов. Эти паттерны помогают повысить производительность и избегать deadlocks.

Worker Pool

Worker Pool - Пул рабочих рук 🤲⚙️ ограниченное число goroutine обрабатывает задачи из канала

Worker pool - кратко - параллельная обработка, позволяет ограничить количество одновременно работающих goroutine, что эффективно для контроля ресурсов.

// Go: Worker Pool
jobs := make(chan int, 10)
results := make(chan int, 10)

for w := 1; w <= 3; w++ {
    go func(id int) {
        for j := range jobs {
            results <- j*2
        }
    }(w)
}

// Отправляем задачи
for j := 1; j <= 5; j++ {
    jobs <- j
}
close(jobs)

// Получаем результаты
for a := 1; a <= 5; a++ {
    fmt.Println(<-results)
}

// Java: Worker Pool
ExecutorService executor = Executors.newFixedThreadPool(3);

try {
    List<Future<Integer>> futures = new ArrayList<>();

    for (int j = 1; j <= 5; j++) {
        int task = j;
        futures.add(executor.submit(() -> task * 2));
    }

    for (Future<Integer> f : futures) {
        System.out.println(f.get());
    }

} finally {
    executor.shutdown();
}

Worker pool полезен, когда нужно ограничить количество одновременно выполняемых задач и контролировать нагрузку на систему.

Pipeline Pattern

Pipeline Pattern - Конвейер обработки данных 🏭➡️ . Данные проходят через последовательные стадии обработки

Pipeline pattern — цепочка стадий, организация последовательной обработки данных через цепочку каналов (stage-by-stage). Позволяет разделять задачи на этапы и эффективно использовать goroutine.

// Go: Pipeline
source := make(chan int, 5)
stage1 := make(chan int, 5)
stage2 := make(chan int, 5)

// Stage 1
go func() {
    for n := range source {
        stage1 <- n * 2
    }
    close(stage1)
}()

// Stage 2
go func() {
    for n := range stage1 {
        stage2 <- n + 1
    }
    close(stage2)
}()

// Отправка данных
for i := 1; i <= 5; i++ {
    source <- i
}
close(source)

// Получение результатов
for r := range stage2 {
    fmt.Println(r)
}

// Java: Pipeline с CompletableFuture
List<CompletableFuture<Integer>> pipeline = new ArrayList<>();

for (int i = 1; i <= 5; i++) {
    pipeline.add(
        CompletableFuture.supplyAsync(() -> i)
            .thenApply(n -> n * 2)
            .thenApply(n -> n + 1)
    );
}

for (CompletableFuture<Integer> f : pipeline) {
    System.out.println(f.get());
}

Pipeline помогает разделять задачи на этапы и снижает вероятность блокировок. Особенно полезно для потоковой обработки данных.

Мини‑схема: «Параллельная обработка задач + сбор результатов»

Визуально паттерны worker pool + fan-in можно показать так:

        jobs
         │
         ▼
   ┌───────────┐
   │  Worker 1 │ (chan)
   └───────────┘
         │
   ┌───────────┐
   │  Worker 2 │ (chan)
   └───────────┘
         │
   ┌───────────┐
   │  Worker 3 │ (chan)
   └───────────┘
         │
         ▼
      results

Livelock

Livelock - Движение без прогресса 🔁 . Потоки активны, но постоянно мешают друг другу

Livelock — это состояние, при котором несколько потоков активно выполняют операции, но не могут продвинуться дальше из-за постоянного взаимодействия между ними. В отличие от deadlock, где потоки полностью заблокированы, при livelock потоки "движутся", но задача не выполняется. Под капотом это часто связано с слишком агрессивными стратегиями повторной попытки (retry), например, когда два потока постоянно уступают друг другу ресурсы, но ни один не получает их в итоге. В Java livelock может возникнуть при неправильной комбинации synchronized блоков и повторных попыток захвата, а в Go — при чрезмерном использовании select с каналами и постоянной переконфигурации состояния.

Пример кода Go/Java

// Go: Пример Livelock с двумя горутинами, которые постоянно уступают друг другу
package main

import (
    "fmt"
    "time"
)

type Fork struct {
    holder string
}

func (f *Fork) Take(name string) bool {
    if f.holder == "" {
        f.holder = name
        return true
    }
    return false
}

func (f *Fork) Release() {
    f.holder = ""
}

func philosopher(name string, left, right *Fork) {
    for {
        // Пытаемся взять левую вилку
        if left.Take(name) {
            fmt.Println(name, "взял левую вилку")
            time.Sleep(10 * time.Millisecond)
            // Пытаемся взять правую вилку
            if right.Take(name) {
                fmt.Println(name, "взял правую вилку и ест")
                right.Release()
                left.Release()
                return
            } else {
                fmt.Println(name, "уступает правой вилке")
                left.Release() // уступаем левую вилку
            }
        }
        time.Sleep(10 * time.Millisecond)
    }
}

func main() {
    fork1 := &Fork{}
    fork2 := &Fork{}

    go philosopher("Альберт", fork1, fork2)
    go philosopher("Боб", fork2, fork1)

    time.Sleep(1 * time.Second)
}
  

// Java: Пример Livelock с двумя потоками, которые постоянно уступают друг другу
class Fork {
    private String holder = "";

    synchronized boolean take(String name) {
        if (holder.isEmpty()) {
            holder = name;
            return true;
        }
        return false;
    }

    synchronized void release() {
        holder = "";
    }
}

public class LivelockExample {
    public static void main(String[] args) throws InterruptedException {
        Fork fork1 = new Fork();
        Fork fork2 = new Fork();

        Runnable philosopherA = () -> {
            try {
                while (true) {
                    if (fork1.take("Альберт")) {
                        Thread.sleep(10);
                        if (fork2.take("Альберт")) {
                            System.out.println("Альберт ест");
                            fork2.release();
                            fork1.release();
                            break;
                        } else {
                            System.out.println("Альберт уступает правой вилке");
                            fork1.release();
                        }
                    }
                    Thread.sleep(10);
                }
            } catch (InterruptedException e) {}
        };

        Runnable philosopherB = () -> {
            try {
                while (true) {
                    if (fork2.take("Боб")) {
                        Thread.sleep(10);
                        if (fork1.take("Боб")) {
                            System.out.println("Боб ест");
                            fork1.release();
                            fork2.release();
                            break;
                        } else {
                            System.out.println("Боб уступает правой вилке");
                            fork2.release();
                        }
                    }
                    Thread.sleep(10);
                }
            } catch (InterruptedException e) {}
        };

        new Thread(philosopherA).start();
        new Thread(philosopherB).start();
    }
}
  

Чтобы избежать livelock, используйте стратегии с экспоненциальной задержкой или случайные паузы между попытками захвата ресурсов. В Go это может быть time.Sleep с разной длительностью, в Java — Thread.sleep с рандомизированным интервалом. Причина: постоянное "подстраивание" потоков друг под друга создаёт бесконечный цикл уступок, который замедляет выполнение. Под капотом JVM или планировщик Go выполняют частые переключения, что дополнительно нагружает CPU.

Livelock часто встречается в распределённых системах или многопоточных приложениях с конфликтующими ресурсами. Например, алгоритмы выбора лидера в распределённых кластерах или ситуации, когда несколько клиентов одновременно пытаются обновить один и тот же объект. Плюсы: активное использование CPU, поток не блокируется полностью. Минусы: отсутствие прогресса и избыточная нагрузка на планировщик. Под капотом видно постоянное переключение контекстов и переоценка состояния ресурсов.

Lock-Free Patterns

Что такое Lock-Free

Lock-Free Patterns - Без блокировок через CAS ⚡ . Потоки работают атомарно без ожидания друг друга

Lock-free паттерны позволяют потокам безопасно работать с общими данными без использования блокировок, используя атомарные операции и сравнение с предыдущим состоянием (CAS — Compare-And-Swap). В Go lock-free достигается через sync/atomic и канализированные структуры, в Java через java.util.concurrent.atomic. Под капотом каждый поток использует атомарные операции CPU, что позволяет избежать блокировок и deadlock, но требует внимательного проектирования, чтобы избежать livelock и ABA-проблемы (когда состояние возвращается к прежнему значению между чтением и записью).

Пример кода Go/Java

// Go: Простейший lock-free счетчик
package main

import (
    "fmt"
    "sync/atomic"
)

func main() {
    var counter int32 = 0

    for i := 0; i < 10; i++ {
        go func() {
            atomic.AddInt32(&counter, 1) // атомарное увеличение
        }()
    }

    fmt.Println("Lock-free счетчик:", atomic.LoadInt32(&counter))
}
  

// Java: Простейший lock-free счетчик
import java.util.concurrent.atomic.AtomicInteger;

public class LockFreeCounter {
    public static void main(String[] args) {
        AtomicInteger counter = new AtomicInteger(0);

        for (int i = 0; i < 10; i++) {
            new Thread(() -> counter.incrementAndGet()).start();
        }

        System.out.println("Lock-free счетчик: " + counter.get());
    }
}
  

Lock-free паттерны эффективны при высокой конкуренции, так как уменьшают накладные расходы на блокировки. Однако важно правильно использовать атомарные операции и учитывать возможные ABA-ситуации, где значение изменилось и вернулось к исходному между чтением и записью. Под капотом CPU обеспечивает атомарность на уровне инструкций, что позволяет избежать взаимных блокировок потоков.

Lock-free структуры широко применяются в высоконагруженных системах, таких как брокеры сообщений, очереди задач, кеши с интенсивным обновлением. Плюсы: высокая производительность, отсутствие deadlock. Минусы: сложность реализации, необходимость правильной работы с памятью и понимания атомарных операций. В Go lock-free структуры часто создаются через каналы и atomic-пакет, в Java через AtomicXXX классы.

Cond (Условные переменные)

Что такое Cond

Cond (Condition Variable) - Сон до сигнала 😴🔔 . Потоки ждут условия и просыпаются по уведомлению

Cond (Condition Variable) — это объект синхронизации, который позволяет одному или нескольким потокам ждать наступления определенного условия и уведомлять других о его выполнении. В Java это java.util.concurrent.locks.Condition, в Go — sync.Cond. Под капотом Cond использует очередь ожидания и сигналы для пробуждения потоков. Это позволяет реализовать сложные схемы синхронизации без постоянного активного опроса состояния, что экономит ресурсы CPU и упрощает дизайн конкурентных алгоритмов.

Пример кода Go/Java

// Go: sync.Cond пример
package main

import (
    "fmt"
    "sync"
)

func main() {
    lock := &sync.Mutex{}
    cond := sync.NewCond(lock)
    ready := false

    go func() {
        lock.Lock()
        for !ready {
            cond.Wait() // ждем сигнала
        }
        fmt.Println("Горутина продолжила выполнение")
        lock.Unlock()
    }()

    lock.Lock()
    ready = true
    cond.Signal() // посылаем сигнал одной горутине
    lock.Unlock()
}
  

// Java: Condition пример
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class CondExample {
    public static void main(String[] args) {
        ReentrantLock lock = new ReentrantLock();
        Condition cond = lock.newCondition();
        boolean[] ready = {false};

        new Thread(() -> {
            lock.lock();
            try {
                while (!ready[0]) {
                    cond.await(); // ждем сигнала
                }
                System.out.println("Поток продолжил выполнение");
            } catch (InterruptedException e) {}
            finally { lock.unlock(); }
        }).start();

        lock.lock();
        try {
            ready[0] = true;
            cond.signal(); // посылаем сигнал
        } finally { lock.unlock(); }
    }
}
  

Использование Cond эффективно для управления зависимостями между потоками без постоянного опроса состояния. Важно правильно обернуть вызовы Wait/Signal внутри блокировки, иначе возможны гонки данных или потеря сигнала. Под капотом очередь ожидания и блокировки реализуют безопасное пробуждение потоков и предотвращение deadlock.

Cond применяется для реализации паттернов Producer-Consumer, очередей задач, синхронизации событий. Плюсы: эффективное ожидание, меньше накладных расходов на CPU. Минусы: требует правильного использования lock/cond комбинаций, иначе легко попасть в deadlock. В Go простая синхронизация через sync.Cond делает код компактнее, в Java чаще используются ReentrantLock + Condition для более гибкого управления потоками.

Лучшие практики

• Ограничивайте количество одновременно работающих goroutine (worker pool).
• Используйте buffered каналы для избежания блокировки при fan-in/fan-out.
• Всегда закрывайте каналы, когда они больше не нужны.
• Мониторьте гонки и deadlocks с помощью go run -race.
• Разделяйте задачи на стадии через pipeline для читаемости и безопасности.

Планирование worker pool и pipeline помогает безопасно масштабировать обработку задач и повышает производительность без риска deadlocks.

Итог

В этой статье мы рассмотрели практические паттерны параллельной обработки задач в Go: worker pool, pipeline pattern и сбор результатов через fan-in. Эти паттерны помогают эффективно использовать ресурсы, избегать блокировок и упрощают масштабирование.

Для Java-разработчика это похоже на использование фиксированных пулов потоков, CompletableFuture и последовательной обработки через stage-by-stage. Освоение этих паттернов позволит создавать высокопроизводительные и безопасные многопоточные приложения.