Goroutines et canaux : modèles de concurrence dans Go

La concurrence nous permet de gérer plusieurs tâches indépendamment les unes des autres. Les Goroutines sont un moyen simple de traiter plusieurs tâches de manière indépendante. Dans cet article, nous améliorons progressivement un gestionnaire http qui accepte les fichiers et explore divers modèles de concurrence dans Go en utilisant les canaux et le package de synchronisation.

Installation

Avant d’entrer dans les modèles de concurrence, préparons le terrain. Imaginez que nous ayons un gestionnaire HTTP qui accepte plusieurs fichiers via un formulaire et traite les fichiers d'une manière ou d'une autre.

func processFile(file multipart.File) {
   // do something with the file
   fmt.Println("Processing file...")
   time.Sleep(100 * time.Millisecond) // Simulating file processing time
}
func UploadHandler(w http.ResponseWriter, r *http.Request) {
   // limit to 10mb 
   if err := r.ParseMultipartForm(10 << 20); err != nil {
       http.Error(w, "Unable to parse form", http.StatusBadRequest)
       return 
   }
   // iterate through all files and process them sequentially 
   for _, file := range r.MultipartForm.File["files"] {
       f, err := file.Open()
       if err != nil {
          http.Error(w, "Unable to read file", http.StatusInternalServerError)
          return
       }
       processFile(f)
       f.Close()
   }
}

Dans l'exemple ci-dessus, nous recevons des fichiers d'un formulaire et les traitons séquentiellement. Si 10 fichiers sont téléchargés, cela prendrait 1 seconde pour terminer le processus et envoyer une réponse au client.
Lors du traitement de nombreux fichiers, cela peut devenir un goulot d'étranglement, mais avec la prise en charge de la concurrence de Go, nous pouvons facilement résoudre ce problème.

Groupes d'attente

Pour résoudre ce problème, nous pouvons traiter les fichiers simultanément. Pour générer une nouvelle goroutine, nous pouvons préfixer un appel de fonction avec le mot-clé go, par exemple. allez processFile (f). Cependant, comme les goroutines ne bloquent pas, le gestionnaire peut revenir avant la fin du processus, laissant les fichiers éventuellement non traités ou renvoyant un état incorrect. Pour attendre le traitement de tous les fichiers, nous pouvons utiliser sync.WaitGroup.
Un WaitGroup attend la fin d'un certain nombre de goroutines, pour chaque goroutine que nous générons, nous devons également augmenter le compteur dans le WaitGroup, cela peut être fait avec la fonction Ajouter. Lorsqu'une goroutine est terminée, Done doit être appelé pour que le compteur soit diminué de un. Avant de revenir de la fonction Wait, il faut appeler ce qui bloque jusqu'à ce que le compteur du WaitGroup soit 0.

func UploadHandler(w http.ResponseWriter, r *http.Request) {
   if err := r.ParseMultipartForm(10 << 20); err != nil {
       http.Error(w, "Unable to parse form", http.StatusBadRequest)
       return
   }

   // create WaitGroup 
   var wg sync.WaitGroup 
   for _, file := range r.MultipartForm.File["files"] {
       f, err := file.Open()
       if err != nil {
          http.Error(w, "Unable to read file", http.StatusInternalServerError)
          return
       }

       wg.Add(1) // Add goroutine to the WaitGroup by incrementing the WaitGroup counter, this should be called before starting a goroutine
       // Process file concurrently
       go func(file multipart.File) {
           defer wg.Done() // decrement the counter by calling Done, utilize defer to guarantee that Done is called. 
           defer file.Close()
           processFile(f)
       }(f)
   }

   // Wait for all goroutines to complete
   wg.Wait()
   fmt.Fprintln(w, "All files processed successfully!")
}

Maintenant, pour chaque fichier téléchargé, une nouvelle goroutine est générée, ce qui pourrait submerger le système. Une solution consiste à limiter le nombre de goroutines générées.

Limiter la concurrence avec un sémaphore

Un sémaphore n'est qu'une variable que nous pouvons utiliser pour contrôler l'accès aux ressources communes par plusieurs threads ou dans ce cas, des goroutines.

En Go, nous pouvons utiliser des canaux tamponnés pour implémenter un sémaphore.

Canaux

Avant de commencer l'implémentation, regardons ce que sont les canaux et la différence entre les canaux avec et sans tampon.

Les canaux sont un canal à travers lequel nous pouvons envoyer et recevoir des données pour communiquer en toute sécurité entre les routines Go.
Les chaînes doivent être créées avec la fonction make.

func processFile(file multipart.File) {
   // do something with the file
   fmt.Println("Processing file...")
   time.Sleep(100 * time.Millisecond) // Simulating file processing time
}
func UploadHandler(w http.ResponseWriter, r *http.Request) {
   // limit to 10mb 
   if err := r.ParseMultipartForm(10 << 20); err != nil {
       http.Error(w, "Unable to parse form", http.StatusBadRequest)
       return 
   }
   // iterate through all files and process them sequentially 
   for _, file := range r.MultipartForm.File["files"] {
       f, err := file.Open()
       if err != nil {
          http.Error(w, "Unable to read file", http.StatusInternalServerError)
          return
       }
       processFile(f)
       f.Close()
   }
}

Les chaînes ont un opérateur spécial <- qui est utilisé pour envoyer ou lire à partir d'une chaîne.
Le fait que l'opérateur pointe sur le canal ch <- 1 envoie des données au canal, si la flèche pointe loin du canal <-ch, la valeur sera reçue. Les opérations d'envoi et de réception sont bloquées par défaut, ce qui signifie que chaque opération attendra que l'autre côté soit prêt.


L'animation visualise un producteur envoyant la valeur 1 via un canal sans tampon et le consommateur lisant depuis le canal.

Si le producteur peut envoyer des événements plus rapidement que le consommateur ne peut les gérer, nous avons la possibilité d'utiliser un canal tamponné pour mettre en file d'attente plusieurs messages sans bloquer le producteur jusqu'à ce que le tampon soit plein. Dans le même temps, le consommateur peut gérer les messages à son propre rythme.

func UploadHandler(w http.ResponseWriter, r *http.Request) {
   if err := r.ParseMultipartForm(10 << 20); err != nil {
       http.Error(w, "Unable to parse form", http.StatusBadRequest)
       return
   }

   // create WaitGroup 
   var wg sync.WaitGroup 
   for _, file := range r.MultipartForm.File["files"] {
       f, err := file.Open()
       if err != nil {
          http.Error(w, "Unable to read file", http.StatusInternalServerError)
          return
       }

       wg.Add(1) // Add goroutine to the WaitGroup by incrementing the WaitGroup counter, this should be called before starting a goroutine
       // Process file concurrently
       go func(file multipart.File) {
           defer wg.Done() // decrement the counter by calling Done, utilize defer to guarantee that Done is called. 
           defer file.Close()
           processFile(f)
       }(f)
   }

   // Wait for all goroutines to complete
   wg.Wait()
   fmt.Fprintln(w, "All files processed successfully!")
}

Dans cet exemple, le producteur peut envoyer jusqu'à deux articles sans bloquer. Lorsque la capacité du tampon est atteinte, le producteur bloquera jusqu'à ce que le consommateur traite au moins un message.

Retour au problème initial, nous voulons limiter le nombre de goroutines traitant des fichiers simultanément. Pour ce faire, nous pouvons utiliser des canaux tamponnés.

ch := make(chan int)

Dans cet exemple nous avons ajouté un canal tamponné d'une capacité de 5, cela nous permet de traiter 5 fichiers simultanément et de limiter la pression sur le système.

Mais et si tous les fichiers ne sont pas égaux ? Nous pouvons prédire de manière fiable que différents types ou tailles de fichiers nécessitent plus de ressources pour être traités. Dans ce cas, nous pouvons utiliser un sémaphore pondéré.

Sémaphore pondéré

En termes simples, avec un sémaphore pondéré, nous pouvons attribuer plus de ressources à une seule tâche. Go fournit déjà une implémentation pour un sémaphore pondéré dans le package extend sync.

ch := make(chan int, 2)

Dans cette version, nous avons créé un sémaphore pondéré avec 5 emplacements. Si seules des images sont téléchargées, par exemple, le processus gère 5 images simultanément. Cependant, si un PDF est téléchargé, 2 emplacements sont acquis, ce qui réduirait la quantité de fichiers pouvant être gérés. en même temps.

Conclusion

Nous avons exploré quelques modèles de concurrence dans Go, en utilisant sync.WaitGroup et des sémaphores pour contrôler le nombre de tâches simultanées. Cependant, il existe davantage d'outils disponibles, nous pourrions utiliser des canaux pour créer un pool de travailleurs, ajouter des délais d'attente ou utiliser un modèle d'entrée/sortie de ventilateur.
De plus, la gestion des erreurs est un aspect important qui a été largement laissé de côté par souci de simplicité.
Une façon de gérer les erreurs consisterait à utiliser des canaux pour regrouper les erreurs et les gérer une fois toutes les goroutines terminées.

Go fournit également un errgroup.Group qui est lié à sync.WaitGroups mais ajoute la gestion des tâches qui renvoient des erreurs.
Le package se trouve dans le package extend sync.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!