Modele de Concurență în Go de Producție: Worker Pools, Pipeline-uri, Fan-Out / Fan-In și Select în Go

Lansarea unei goroutine noi prin cuvântul cheie go este atât de facilă și ieftină încât o greșeală comună a dezvoltatorilor începători este pornirea necontrolată a unei goroutine pentru fiecare element primit (de exemplu, o goroutine per rând dintr-un fișier de 10 milioane de înregistrări sau per cerere HTTP). Această abordare naivă duce rapid la epuizarea descriptorilor de fișiere („Too many open files”), saturarea conexiunilor la baza de date și prăbușirea serverului prin epuizarea memoriei RAM.

1. Modelul Worker Pool: Limitarea Strictă a Concurenței

Un Worker Pool menține un număr fix de goroutine-uri lucrătoare (de exemplu, un număr egal cu numărul de nuclee CPU sau cu numărul maxim de conexiuni permise în baza de date) care consumă sarcini dintr-un canal comun de intrare cu buffer:

package main

import (
    "fmt"
    "sync"
    "time"
)

type Sarcina struct {
    ID     int
    Date   string
}

type Rezultat struct {
    SarcinaID int
    Iesire    string
}

// Funcția executată de fiecare worker în fundal:
func worker(id int, canalSarcini <-chan Sarcina, canalRezultate chan<- Rezultat, wg *sync.WaitGroup) {
    defer wg.Done()
    for s := range canalSarcini {
        // Simulăm procesarea sarcinii:
        time.Sleep(50 * time.Millisecond)
        canalRezultate <- Rezultat{
            SarcinaID: s.ID,
            Iesire:    fmt.Sprintf("Worker-ul %d a procesat sarcina %d", id, s.ID),
        }
    }
}

func main() {
    const NUMAR_WORKERI = 4
    const TOTAL_SARCINI = 12

    canalSarcini := make(chan Sarcina, TOTAL_SARCINI)
    canalRezultate := make(chan Rezultat, TOTAL_SARCINI)

    var wg sync.WaitGroup

    // 1. Pornim grupul fix de workeri:
    for w := 1; w <= NUMAR_WORKERI; w++ {
        wg.Add(1)
        go worker(w, canalSarcini, canalRezultate, &wg)
    }

    // 2. Trimitem sarcinile în canal și închidem canalul când am terminat:
    for j := 1; j <= TOTAL_SARCINI; j++ {
        canalSarcini <- Sarcina{ID: j, Date: fmt.Sprintf("Date-%d", j)}
    }
    close(canalSarcini) // Semnalizează workerilor că nu mai există sarcini noi!

    // 3. Așteptăm finalizarea workerilor într-o goroutine separată și închidem rezultatele:
    go func() {
        wg.Wait()
        close(canalRezultate)
    }()

    // 4. Colectăm toate rezultatele:
    for rez := range canalRezultate {
        fmt.Println(rez.Iesire)
    }
}

2. Prevenirea Scurgerilor de Goroutine-uri (Goroutine Leaks)

O scurgere de goroutine apare atunci când o goroutine este blocată pentru totdeauna așteptând să citească sau să scrie pe un canal care nu va mai fi accesat niciodată. Deoarece goroutine-ul nu poate fi curățat de Garbage Collector în timp ce este activ, memoria stivei sale rămâne alocată pe vecie:

• Regula #1: Asigurați-vă întotdeauna că o goroutine are o cale clară de ieșire prin ascultarea pe un canal de context: select { case <-ctx.Done(): return }.
• Regula #2: Pentru canalele prin care trimiteți doar un singur rezultat către o goroutine ce poate abandona apelul (din cauza unui timeout), folosiți un canal cu buffer de mărime 1 (ch := make(chan int, 1)). Astfel, goroutine-ul trimițător poate scrie rezultatul fără a se bloca, chiar dacă celălalt capăt a ieșit deja din funcție!