Fan-Out/Fan-In en Go : motif pour le traitement parallèle des données

Motif Fan-Out/Fan-In en Go : Traitement parallèle flexible des données

Le motif Fan-Out/Fan-In offre une méthode puissante en Go pour répartir les tâches sur plusieurs goroutines et agréger ensuite les résultats. Contrairement au motif Worker Pool, il apporte une plus grande flexibilité grâce à des pipelines de traitement individuels pour chaque worker, ce qui est particulièrement utile pour des chaînes de transformation de données complexes.

Principes fondamentaux du motif

Le motif se décompose en deux phases principales. La phase Fan-Out répartit les données d'entrée sur plusieurs workers (goroutines). Chaque worker reçoit sa propre tâche, la traite de manière indépendante et envoie le résultat vers son canal dédié. La phase Fan-In collecte ensuite les résultats de tous ces canaux individuels dans un seul canal de sortie pour le traitement aval.

La principale différence avec le Worker Pool réside dans l'architecture des canaux :

• Worker Pool : Tous les workers lisent généralement les tâches depuis un canal partagé et écrivent les résultats dans un autre canal partagé.
• Fan-Out/Fan-In : Chaque worker possède son propre canal de sortie, nécessitant une étape supplémentaire de fusion.

Cette approche ajoute un léger surcoût dû aux goroutines et canaux supplémentaires pour la fusion, mais elle excelle dans les architectures en pipeline où les étapes ultérieures nécessitent des flux de données séparés.

Implémentation pratique

Parcourons une implémentation concrète, délibérément différente d'un Worker Pool classique pour mettre en lumière les caractéristiques clés.

Fonction worker

Dans cette configuration, le worker est une fonction qui lance une goroutine et retourne un canal de résultats. Cela diffère du passage d'un canal en paramètre.

func worker(x int, workerID int) chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        fmt.Printf("worker %d, job %d\n", workerID, x)
        out <- x * x
    }()
    return out
}

Détails d'implémentation clés :

• La goroutine ferme son canal out avec defer après envoi du résultat.
• Pour la démo, le travail est trivial — élévation au carré d'un nombre.
• En code de production, il faudrait affiner : gérer le cycle de vie avec un context pour des arrêts gracieux et des timeouts.

Phase Fan-Out (Répartition)

Le distributeur (producteur) lance un worker par élément de données d'entrée. Les canaux retournés par les workers sont stockés dans une tranche pour une fusion ultérieure.

for i, x := range data {
    chnls = append(chnls, worker(x, i+1))
}

Caractéristique clé : Le nombre de workers s'adapte dynamiquement au nombre de tâches, ce qui n'est pas obligatoire mais illustre la flexibilité du motif. Dans les applications réelles, on limite généralement le nombre de workers concurrents.

Phase Fan-In et fonction de fusion

La fonction merge gère la tâche cruciale de combiner les données de plusieurs canaux en un seul. Elle :

• Crée un canal de sortie unique non tamponné out.
• Pour chaque canal d'entrée, lance une goroutine qui lit les données et les transfère vers out.
• Dans une goroutine séparée, attend la fin de toutes les goroutines lectrices avec sync.WaitGroup puis ferme out.

func merge(chnls ...chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    for _, ch := range chnls {
        wg.Add(1)
        go func(c chan int) {
            defer wg.Done()
            for x := range c {
                out <- x
            }
        }(ch)
    }

    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

Note technique : L'attente (wg.Wait()) s'exécute dans une goroutine séparée. La faire dans le flux principal pourrait causer un blocage si le consommateur n'a pas encore commencé à lire out, bloquant les écritures.

Consommateur

Le consommateur lit les résultats du canal fusionné unique. La boucle for range se termine automatiquement quand merge ferme out.

for x := range out {
    results = append(results, x)
}

Points clés à retenir

• Flexibilité architecturale : Des pipelines individuels par worker facilitent l'insertion d'étapes de traitement supplémentaires.
• Évolutivité dynamique : Théoriquement, un worker par tâche, mais on contrôle la concurrence en pratique.
• Surcoût : Le motif consomme des ressources supplémentaires (canaux, goroutines) pour la fusion par rapport à des approches plus simples.
• Gestion du cycle de vie : Cette démo omet les context, acceptable pour des exemples mais essentiel en production pour les annulations et timeouts.
• Lisibilité du code : Les canaux non tamponnés et boucles for range gardent le code concis et clair.

Recommandations pour la production

Ce code de démo est éducatif. Pour un usage réel, appliquez ces améliorations :

• Ajouter un context (context.Context) : Le passer aux workers et à merge pour gérer annulations, deadlines et timeouts.
• Limiter la concurrence : Des goroutines illimitées (une par tâche) épuisent les ressources. Utilisez des sémaphones ou un pool de workers à taille fixe en Fan-Out.
• Gestion d'erreurs : Prévoyez une propagation des erreurs des workers vers le consommateur, via un canal d'erreurs dédié ou une struct de résultats.
• Canaux tamponnés : Ils boostent les performances en réduisant les blocages dans certains cas, mais testez minutieusement.

Le motif Fan-Out/Fan-In est un outil incontournable dans l'arsenal de tout développeur Go pour construire des pipelines de traitement de données efficaces et évolutifs — surtout quand une grande flexibilité est nécessaire pour gérer les flux entre étapes.

— Editorial Team