Какие паттерны конкурентности вы знаете в Go?

Основные паттерны конкурентности в Go

Go, благодаря своей философии «конкурентность не равна параллельности» и наличию мощных первоклассных (first-class) инструментов, таких как goroutines, channels и select, предлагает уникальные и эффективные паттерны для работы с многозадачностью. Эти паттерны часто отличаются от классических подходов в других языках, так как они построены вокруг коммуникации и синхронизации через каналы, а не разделяемой памяти и мьютексов.

1. Worker Pool (Пул рабочих goroutines)

Это классический паттерн для управления пулом ограниченного количества goroutines, которые обрабатывают задачи из общего канала. Он предотвращает создание бесконечного числа goroutines и позволяет контролировать нагрузку.

func worker(id int, jobs <-chan int, results chan<- int) {
    for job := range jobs {
        results <- job * 2 // Пример обработки
    }
}

func main() {
    jobs := make(chan int, 100)
    results := make(chan int, 100)

    // Создаем пул из 3 рабочих
    for w := 1; w <= 3; w++ {
        go worker(w, jobs, results)
    }

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

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

2. Fan-out / Fan-in (Разделение и объединение потоков)

• Fan-out: несколько goroutines («воркеры») читают из одного входного канала, параллельно обрабатывая данные.
• Fan-in: одна goroutine объединяет результаты из нескольких каналов в один выходной канал, часто используя select.

Этот паттерн идеален для распараллеливания этапа обработки и последующей агрегации результатов.

// Fan-out: несколько обработчиков читают из inCh
// Fan-in: merge объединяет результаты из нескольких каналов в outCh
func merge(outCh chan<- int, inChs ...<-chan int) {
    var wg sync.WaitGroup
    wg.Add(len(inChs))

    for _, inCh := range inChs {
        go func(ch <-chan int) {
            for v := range ch {
                outCh <- v
            }
            wg.Done()
        }(inCh)
    }
    wg.Wait()
    close(outCh)
}

3. Generator (Генератор)

Функция, которая возвращает канал и запускает внутреннюю goroutine, отправляющую значения в этот канал. Позволяет абстрагировать процесс производства данных.

func countGen(start, end int) <-chan int {
    ch := make(chan int)
    go func() {
        for i := start; i <= end; i++ {
            ch <- i
        }
        close(ch)
    }()
    return ch // Возвращаем канал только для чтения
}

// Использование: for num := range countGen(1, 5) { ... }

4. Pipeline (Конвейер)

Комбинация нескольких этапов обработки, где каждый этап представлен goroutine, получающей данные из входного канала и отправляющей результат в выходной канал. Этапы соединены каналами, образуя конвейер данных.

func stageMultiply(in <-chan int, factor int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            out <- n * factor
        }
        close(out)
    }()
    return out
}

func main() {
    nums := countGen(1, 5)          // Генератор
    doubled := stageMultiply(nums, 2) // Первый этап конвейера
    squared := stageMultiply(doubled, 2) // Второй этап (условно)
    for res := range squared {
        fmt.Println(res)
    }
}

5. Done Channel (Канал для завершения)

Канал (done chan struct{}), используемый для сигнализации goroutines о необходимости прекратить работу. Часто используется вместе с select для обеспечения корректного и безопасного завершения («graceful shutdown»).

func worker(done <-chan struct{}, dataCh <-chan int) {
    for {
        select {
        case data := <-dataCh:
            process(data)
        case <-done: // Получен сигнал завершения
            fmt.Println("Worker stopping")
            return
        }
    }
}

func main() {
    done := make(chan struct{})
    dataCh := make(chan int)
    go worker(done, dataCh)

    // ... работа системы ...

    // Сигнал всем goroutines о завершении
    close(done)
}

6. Semaphore Pattern (Семафор через канал с буфером)

Канал с ограниченной емкостью (буфером) используется как семафор для контроля количества одновременно выполняющихся операций или доступа к ресурсу.

var sem = make(chan struct{}, 3) // Семафор на 3 "токена"

func accessResource(id int) {
    sem <- struct{}{}          // Занимаем токен (блокируемся если нет свободных)
    defer func() { <-sem }()   // Освобождаем токен при выходе

    fmt.Printf("Resource accessed by %d\n", id)
    time.Sleep(time.Second)
}

7. Pub/Sub (Publisher/Subscriber) через каналы

Хотя для сложных систем лучше использовать готовые библиотеки, базовый паттерн можно реализовать, где «публикатор» отправляет сообщения в центральный канал, а несколько «подписчиков» (goroutines) получают их копии через отдельные каналы, создаваемые при регистрации.

8. Barrier Synchronization (Барьер для синхронизации)

Синхронизация нескольких goroutines в одной точке, часто реализуемая через sync.WaitGroup или канал, который блокирует всех до выполнения определенного условия.

9. Context для управления жизненным циклом

Использование context.Context — это не просто паттерн, а стандартный механизм Go для передачи сигналов завершения, deadlines и значений через цепочки вызовов и goroutines. Он интегрируется с select и каналами для управления конкурентными операциями.

Ключевые принципы и отличие от других языков

• Коммуникация через каналы (Do not communicate by sharing memory; share memory by communicating): Это основная идея Go. Вместо использования мьютексов для защиты разделяемой памяти, данные передаются между goroutines через каналы. Это снижает риск race conditions.
• Select как мультиплексор: Оператор select позволяет goroutine ждать на нескольких каналах одновременно, что является фундаментом для многих сложных паттернов (fan-in, done channel).
• Легковесные goroutines: Паттерны часто предполагают создание множества goroutines, так как они экономичны по ресурсам.

Эти паттерны образуют основу для построения эффективных, безопасных и читаемых конкурентных программ в Go. Выбор паттерна зависит от конкретной задачи: обработка потоков данных (pipeline, fan-out/fan-in), управление задачами (worker pool), контроль исполнения (done channel, context) или ограничение ресурсов (semaphore).