Problèmes d'application complets des coroutines Go et des canaux

Problèmes d'application complets des coroutines et des canaux Go

2024-07-12

1. Comprendre brièvement ce que sont les coroutines et les canaux

Qu'est-ce que la coroutine

Une coroutine est un thread léger au niveau utilisateur qui dispose d'un espace de pile indépendant et partage l'espace de tas du programme.

Il s'agit d'un micro-thread implémenté via des algorithmes basés sur un seul thread. Par rapport à la programmation multi-thread, il présente les avantages suivants :

Le changement de contexte de la coroutine est décidé par l'utilisateur, sans qu'il soit nécessaire de changer de contexte du noyau système, ce qui réduit la surcharge.
Par défaut, les coroutines sont entièrement protégées pour éviter les interruptions.Aucun verrouillage de fonctionnement atomique requis
Un seul thread peut également atteindre une concurrence élevée, et même un processeur monocœur peut prendre en charge des dizaines de milliers de coroutines.

qu'est-ce que la chaîne

Le canal est une structure de données utilisée pour la communication entre les coroutines. Semblable à une file d’attente, une extrémité est l’expéditeur et l’autre extrémité est le destinataire. L'utilisation de canaux peut garantir la synchronisation et l'ordre des données.

Les canaux sont divisés en canaux tamponnés et canaux non tamponnés, déclarés comme suit :

Il y a un canal tampon

intChan := make(chan int,<缓冲容量>)

canal sans tampon

intChan := make(chan int)

La différence entre les canaux tamponnés et les canaux non tamponnés :

Blocage : l'expéditeur d'un canal sans tampon bloquera jusqu'à ce que les données soient reçues ; l'expéditeur d'un canal avec tampon bloquera jusqu'à ce que le tampon soit plein, et le récepteur bloquera jusqu'à ce que le tampon ne soit pas vide.
Synchronisation et séquence des données : les canaux sans tampon garantissent la synchronisation et la séquence des données ; les canaux mis en mémoire tampon ne garantissent pas la synchronisation et la séquence des données.
Scénarios d'application : les canaux sans tampon nécessitent une synchronisation et une séquence strictes ; les canaux avec tampon peuvent communiquer de manière asynchrone et améliorer le débit.

Ce qu'il faut noter dans la mise en œuvre de canaux sans tampon, c'est qu'il doit y avoir un émetteur et un récepteur aux deux extrémités du canal, sinon un blocage se produira.

2. Cas de programmation simultanée par canal Coroutine

(1) Imprimez alternativement les lettres et les chiffres

Signification de la question : utilisez le canal coroutine pour imprimer alternativement les chiffres 1 à 10 et les lettres AJ.

Code:


package main
 
import (
	"fmt"
	"sync"
)
 
/*
无缓冲chanel：需要在写入chanel的时候要保证有另外一个协程在读取chanel。否则会导致写端阻塞，发生死锁
解决办法：
避免死锁的发生：
当i循环到10时，printAlp协程已然结束，所以此时不必再写入alp通道
*/
 
func printNum(wg *sync.WaitGroup, numCh chan struct{}, alpCh chan struct{}) {
	defer wg.Done()
 
	for i := 1; i <= 10; i++ {
		<-alpCh // 等待字母goroutine发信号
		fmt.Print(i, " ")
		//避免死锁发生
		if i < 10 {
			numCh <- struct{}{} // 发信号给字母goroutine
		}
		if i == 10 {
			close(numCh)
		}
	}
 
}
 
func printAlp(wg *sync.WaitGroup, numCh chan struct{}, alpCh chan struct{}) {
	defer wg.Done()
 
	for i := 'A'; i <= 'J'; i++ {
		<-numCh // 等待数字goroutine发信号
		fmt.Printf("%c", i)
		alpCh <- struct{}{} // 发信号给数字goroutine
	}
	close(alpCh)
}
 
func main() {
	numCh := make(chan struct{}) // 用于数字goroutine的信号通道
	alpCh := make(chan struct{}) // 用于字母goroutine的信号通道
	var wg sync.WaitGroup
 
	wg.Add(2)
 
	go printAlp(&wg, numCh, alpCh)
	go printNum(&wg, numCh, alpCh)
 
	// 启动时先给数字goroutine发送一个信号
	numCh <- struct{}{}
 
	wg.Wait()
 
}

Analyse du sujet :

La question nous oblige à imprimer alternativement des lettres et des chiffres, nous devons donc garantir l'ordre strict des deux coroutines, ce qui est cohérent avec le scénario d'application des canaux sans tampon. Configurez deux canaux pour stocker respectivement des chiffres et des lettres. Deux coroutines qui impriment des chiffres et des lettres servent respectivement d'expéditeur et de récepteur des deux canaux. Imprimez une fois en boucle et envoyez un signal une fois pour rappeler à une autre coroutine d'imprimer.

Il convient de noter que lorsque le dernier caractère « 10 » est imprimé, la coroutine d'impression des lettres est terminée et le canal numCh n'a pas de récepteur. A ce moment, les conditions de mise en œuvre du canal non tamponné ne sont plus remplies - il doit y en avoir. un émetteur et un récepteur, sinon l'envoi à nouveau du signal entraînera un blocage. Il n’est donc pas nécessaire de renvoyer le signal une dixième fois.

(2) Concevoir un planificateur de tâches

Titre : Concevoir un planificateur de tâches qui utilise le modèle de programmation multi-coroutine + canal pour mettre en œuvre des scénarios commerciaux de traitement simultané de multi-tâches, et nécessite que l'ordre de planification soit dans l'ordre dans lequel les tâches sont ajoutées.

Code:


type scheduler struct {
	taskChan chan func()
	wt       sync.WaitGroup
}
 
func (td *scheduler) AddTask(task func()) {
	td.taskChan <- task
}
 
func (td *scheduler) Executer() {
	defer td.wt.Done()
	for {
		task, ok := <-td.taskChan
		task()
		if ok && len(td.taskChan) == 0 {
			break
		}
	}
}
 
func (td *scheduler) Start() {
	td.wt.Add(4)
	//假设四个消费者
	for i := 0; i < 4; i++ {
		go td.Executer()
	}
 
	td.wt.Wait()
}
 
func main() {
	sd := scheduler{
		taskChan: make(chan func(), 5),
	}
 
	go func() {
		sd.AddTask(func() {
			fmt.Println("任务1")
		})
		sd.AddTask(func() {
			fmt.Println("任务2")
		})
		sd.AddTask(func() {
			fmt.Println("任务3")
		})
		sd.AddTask(func() {
			fmt.Println("任务4")
		})
		sd.AddTask(func() {
			fmt.Println("任务5")
		})
		sd.AddTask(func() {
			fmt.Println("任务6")
		})
		close(sd.taskChan)
	}()
 
	sd.Start()
 
}

analyse du problème:

Étant donné que les tâches ajoutées sont multitâches, il y en a plusieurs et un traitement asynchrone est requis pour exécuter ces tâches. Le respect des canaux mis en mémoire tampon nécessite un débit amélioré et un traitement asynchrone.

Ensuite, nous devons placer la tâche dans le canal, et plusieurs récepteurs peuvent prendre les tâches du canal dans l'ordre et les exécuter.

Le problème auquel il faut prêter attention est que si le nombre de tâches ajoutées est supérieur au tampon du canal, cela provoquera un blocage lors de l'ajout de tâches. Afin de ne pas affecter le démarrage normal du consommateur, celui-ci doit ouvrir une coroutine distincte pour ajouter des tâches.

De cette façon, lorsque le consommateur consommera, le producteur bloquant sera réveillé pour continuer à ajouter des tâches.

3. Résumé

Après avoir étudié le modèle de programmation coroutine + canal, en plus de ce qui vient d'être mentionné dans le titre, nous devons également prêter attention aux problèmes suivants :

1. Pourquoi le canal devrait-il être fermé une fois épuisé ? Quels sont les risques de ne pas le fermer ?

Pour éviter une impasse. La fermeture du canal indique également au récepteur qu'il n'y a plus de données à envoyer de la part de l'expéditeur et qu'il n'est pas nécessaire de continuer à attendre les données. Après avoir reçu l'information de fermeture du canal, le récepteur cesse de recevoir des données ; si le canal n'est pas fermé, le récepteur restera bloqué et il existe un risque de blocage.
Libérez des ressources et évitez les fuites de ressources. Après avoir fermé le canal, le système libérera les ressources correspondantes. La fermeture du canal à temps peut éviter le gaspillage et les fuites de ressources.

2. Comment fermer la chaîne en douceur ?

Tout d’abord, le principe le plus fondamental de la fermeture des chaînes est de ne pas fermer les chaînes qui ont été fermées. Deuxièmement, il existe un autre principe d'utilisation des canaux Go :Ne fermez pas le canal sur le récepteur de données ou lorsqu'il y a plusieurs expéditeurs.autrement dit，Nous devrions uniquement laisser le seul expéditeur d'un canal fermer ce canal.

Une méthode grossière consiste à fermer le canal via une récupération d'exception, mais cela viole évidemment les principes ci-dessus et peut provoquer une concurrence en matière de données. Une autre méthode consiste à fermer le canal avec sync.Once ou sync.Mutex, ce qui n'est pas garanti. les opérations et les opérations d'envoi sur un canal ne créent pas de courses de données. Les deux méthodes présentent certains problèmes, je ne les présenterai donc pas en détail. Voici une méthode permettant de fermer le canal en douceur.

Scénario 1 : M récepteurs et un expéditeur

Une des situations les plus faciles à gérer. Lorsque l'expéditeur doit terminer l'envoi, laissez-le simplement fermer le canal. C'est le cas dans les deux exemples de programmation ci-dessus.

Scénario 2 : un récepteur et N expéditeurs

Selon les principes de base des chaînes Go, nous ne pouvons fermer la chaîne qu'au niveau du seul émetteur de la chaîne. Donc, dans ce cas, nous ne pouvons pas fermer directement le canal quelque part.Mais nous pouvons laisser le récepteur fermer un canal de signal supplémentaire pour indiquer à l'expéditeur de ne plus envoyer de données.。


package main
 
import (
	"log"
	"sync"
)
 
func main() {
 
	cosnt N := 5
	cosnt Max := 60000
	count := 0
 
	dataCh := make(chan int)
	stopCh := make(chan bool)
 
	var wt sync.WaitGroup
	wt.Add(1)
 
	//发送者
	for i := 0; i < N; i++ {
		go func() {
			for {
				select {
				case <-stopCh:
					return
				default:
					count += 1
					dataCh <- count
				}
			}
		}()
	}
 
	//接收者
	go func() {
		defer wt.Done()
		for value := range dataCh {
			if value == Max {
				// 此唯一的接收者同时也是stopCh通道的
				// 唯一发送者。尽管它不能安全地关闭dataCh数
				// 据通道，但它可以安全地关闭stopCh通道。
				close(stopCh)
				return
			}
			log.Println(value)
		}
	}()
 
	wt.Wait()
}

Dans cette méthode, nous ajoutons un canal de signal supplémentaire stopCh, que le récepteur utilise pour indiquer à l'expéditeur qu'il n'a plus besoin de recevoir de données. De plus, cette méthode ne ferme pas dataCh Lorsqu'un canal n'est plus utilisé par aucune coroutine, il sera progressivement récupéré, qu'il ait été fermé ou non.

L'élégance de cette méthode est qu'en fermant un canal, vous arrêtez d'utiliser un autre canal, fermant ainsi indirectement l'autre canal.

Scénario 3 : M récepteurs et N émetteurs

Nous ne pouvons pas demander au récepteur ou à l'expéditeur de fermer le canal utilisé pour transmettre les données, ni à l'un des multiples récepteurs de fermer un canal de signalisation supplémentaire. Ces deux pratiques violent le principe de fermeture des canaux.

On peut cependant introduireUn rôle de médiateur intermédiaire et lui faire fermer des canaux de signalisation supplémentaires pour notifier tous les destinataires et expéditeurs de la fin des travaux。

Exemple de code :


package main
 
import (
	"log"
	"math/rand"
	"strconv"
	"sync"
)
 
func main() {
 
	const Max = 100000
	const NumReceivers = 10
	const NumSenders = 1000
 
	var wt sync.WaitGroup
	wt.Add(NumReceivers)
 
	dataCh := make(chan int)
	stopCh := make(chan struct{})
	// stopCh是一个额外的信号通道。它的发送
	// 者为中间调解者。它的接收者为dataCh
	// 数据通道的所有的发送者和接收者。
	toStop := make(chan string, 1)
	// toStop是一个用来通知中间调解者让其
	// 关闭信号通道stopCh的第二个信号通道。
	// 此第二个信号通道的发送者为dataCh数据
	// 通道的所有的发送者和接收者，它的接收者
	// 为中间调解者。它必须为一个缓冲通道。
 
	var stoppedBy string
 
	// 中间调解者
	go func() {
		stoppedBy = <-toStop
		close(stopCh)
	}()
 
	// 发送者
	for i := 0; i < NumSenders; i++ {
		go func(id string) {
			for {
				value := rand.Intn(Max)
				if value == 0 {
					// 为了防止阻塞，这里使用了一个尝试
					// 发送操作来向中间调解者发送信号。
					select {
					case toStop <- "发送者#" + id:
					default:
					}
					return
				}
 
				select {
				case <-stopCh:
					return
				case dataCh <- value:
				}
			}
		}(strconv.Itoa(i))
	}
 
	// 接收者
	for i := 0; i < NumReceivers; i++ {
		go func(id string) {
			defer wt.Done()
 
			for {
				select {
				case <-stopCh:
					return
				case value := <-dataCh:
					if value == Max {
						// 为了防止阻塞，这里使用了一个尝试
						// 发送操作来向中间调解者发送信号。
						select {
						case toStop <- "接收者:" + id:
						default:
						}
						return
					}
 
					log.Println(value)
				}
			}
		}(strconv.Itoa(i))
	}
 
	wt.Wait()
	log.Println("被" + stoppedBy + "终止了")
 
}

Partage de technologie