Problemi applicativi completi di Go coroutines e channel

Problemi applicativi completi delle coroutine e dei canali Go

2024-07-12

1. Comprendi brevemente cosa sono le coroutine e i canali

Cos'è la coroutine

Una coroutine è un thread leggero a livello utente che ha uno spazio stack indipendente e condivide lo spazio heap del programma.

È un micro-thread implementato tramite algoritmi sulla base del single thread. Rispetto alla programmazione multi-thread presenta i seguenti vantaggi:

Il cambio di contesto della coroutine viene deciso dall'utente, senza la necessità di cambiare contesto del kernel del sistema, riducendo il sovraccarico.
Per impostazione predefinita, le coroutine sono completamente protette per evitare interruzioni.Non è richiesto alcun blocco del funzionamento atomico
Un singolo thread può anche raggiungere un'elevata concorrenza e persino una CPU single-core può supportare decine di migliaia di coroutine.

cos'è il canale

Il canale è una struttura dati utilizzata per la comunicazione tra coroutine. Simile a una coda, un'estremità è il mittente e l'altra estremità è il destinatario. L'utilizzo dei canali può garantire la sincronizzazione e l'ordine dei dati.

I canali sono divisi in canali con buffer e canali senza buffer, che sono dichiarati come segue:

C'è un canale buffer

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

canale senza buffer

intChan := make(chan int)

La differenza tra canali bufferizzati e canali senza buffer:

Blocco: il mittente di un canale senza buffer si bloccherà finché i dati non verranno ricevuti; il mittente di un canale con buffer si bloccherà finché il buffer non sarà pieno e il destinatario si bloccherà finché il buffer non sarà vuoto.
Sincronizzazione e sequenza dei dati: i canali senza buffer garantiscono la sincronizzazione e la sequenza dei dati; le pipe bufferizzate non garantiscono la sincronizzazione e la sequenza dei dati.
Scenari applicativi: i canali senza buffer richiedono una sincronizzazione e una sequenza rigorose; i canali con buffer possono comunicare in modo asincrono e migliorare la velocità effettiva.

Ciò che è necessario notare nell'implementazione dei canali senza buffer è che devono esserci un mittente e un destinatario su entrambe le estremità del canale, altrimenti si verificherà una situazione di stallo.

2. Caso di programmazione simultanea di canali coroutine

(1) Stampa alternativamente lettere e numeri

Significato della domanda: utilizzare il canale coroutine per stampare alternativamente i numeri 1-10 e le lettere AJ.

Codice:


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()
 
}

Analisi dell'argomento:

La domanda richiede di stampare alternativamente lettere e numeri, quindi dobbiamo garantire l'ordine rigoroso delle due coroutine, che sia coerente con lo scenario applicativo dei canali senza buffer. Imposta due canali per memorizzare rispettivamente numeri e lettere. Due coroutine che stampano numeri e lettere fungono rispettivamente da mittente e destinatario. Stampa una volta in un ciclo e invia un segnale una volta per ricordare a un'altra coroutine di stampare.

Va notato che quando viene stampato l'ultimo carattere '10', la coroutine per la stampa delle lettere è terminata e il canale numCh non ha alcun ricevitore. A questo punto, le condizioni di implementazione del canale senza buffer non sono più soddisfatte - devono esserci un mittente e un destinatario In caso contrario, l'invio successivo del segnale causerà un deadlock bloccante. Quindi non è necessario inviare nuovamente il segnale per la decima volta.

(2) Progettare un pianificatore di attività

Titolo: Progettare un'utilità di pianificazione delle attività che utilizza il modello di programmazione multi-coroutine + canale per implementare scenari aziendali di elaborazione simultanea di più attività e richiede che l'ordine di pianificazione sia nell'ordine in cui vengono aggiunte le attività.

Codice:


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()
 
}

Analisi del problema:

Poiché le attività aggiunte sono multi-task, ce n'è più di una e per eseguire queste attività è necessaria l'elaborazione asincrona. La conformità ai canali bufferizzati richiede un throughput migliorato e un'elaborazione asincrona.

Quindi, dobbiamo inserire l'attività nel canale e più ricevitori possono prendere le attività dal canale in ordine ed eseguirle.

Il problema che richiede attenzione è che se il numero di attività aggiunte è maggiore del buffer del canale, ciò causerà un blocco durante l'aggiunta di attività. Per non influenzare il normale avvio del consumer, è necessario aprire una coroutine separata per aggiungere attività.

In questo modo, quando il consumatore consuma, il produttore bloccante verrà risvegliato per continuare ad aggiungere compiti.

3. Riepilogo

Dopo aver studiato il modello di programmazione coroutine + canale, oltre a quanto appena accennato nel titolo, dovremmo prestare attenzione anche alle seguenti questioni:

1. Perché il canale dovrebbe essere chiuso una volta esaurito. Quali sono i rischi se non lo si chiude?

Per evitare lo stallo. La chiusura del canale indica inoltre al destinatario che non ci sono più dati da inviare dal mittente e non è necessario continuare ad attendere i dati. Dopo aver ricevuto l'informazione di chiusura del canale, il ricevitore smette di ricevere i dati; se il canale non viene chiuso, il ricevitore rimarrà bloccato e c'è il rischio di stallo.
Liberare risorse ed evitare perdite di risorse. Dopo aver chiuso il canale, il sistema rilascerà le risorse corrispondenti. La chiusura del canale in tempo può evitare sprechi e perdite di risorse.

2. Come chiudere il canale con garbo?

Innanzitutto, il principio fondamentale per chiudere i canali è non chiudere i canali che sono stati chiusi. In secondo luogo, esiste un altro principio per l'utilizzo dei canali Go:Non chiudere il canale sul ricevitore dati o quando sono presenti più mittenti.in altre parole，Dovremmo consentire solo all'unico mittente di un canale di chiudere questo canale.

Un modo scortese è chiudere il canale tramite il ripristino dell'eccezione, ma ciò ovviamente viola i principi di cui sopra e può causare una corsa ai dati. Un altro modo è chiudere il canale con sync.Once o sync.Mutex, cosa che non è garantita che avvenga le operazioni e le operazioni di invio su un canale non creano gare di dati. Entrambi i metodi presentano alcuni problemi, quindi non li presenterò in dettaglio. Ecco un metodo su come chiudere con grazia il canale.

Scenario 1: M ricevitori e un mittente

Una delle situazioni più facili da affrontare. Quando il mittente deve terminare l'invio, lascia che chiuda il canale. Questo è il caso dei due esempi di programmazione sopra.

Scenario 2: un destinatario e N mittenti

Secondo i principi di base dei canali Go, possiamo chiudere il canale solo presso l'unico mittente del canale. Quindi, in questo caso, non possiamo chiudere direttamente il canale da qualche parte.Ma possiamo lasciare che il ricevitore chiuda un canale di segnale aggiuntivo per dire al mittente di non inviare altri dati.。


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()
}

In questo metodo aggiungiamo un ulteriore canale di segnale stopCh, che il ricevitore utilizza per comunicare al mittente che non ha più bisogno di ricevere dati. Inoltre, questo metodo non chiude dataCh Quando un canale non viene più utilizzato da nessuna coroutine, verrà gradualmente sottoposto a garbage collection, indipendentemente dal fatto che sia stato chiuso.

L'eleganza di questo metodo è che chiudendo un canale si smette di utilizzare un altro canale, chiudendo così indirettamente l'altro canale.

Scenario 3: M ricevitori e N mittenti

Non possiamo fare in modo che né il ricevitore né il mittente chiudano il canale utilizzato per trasmettere i dati, né possiamo fare in modo che uno dei molteplici ricevitori chiuda un canale di segnalazione aggiuntivo. Entrambe queste pratiche violano il principio di chiusura del canale.

Possiamo però presentarciUn ruolo di mediatore intermedio e la possibilità di chiudere ulteriori canali di segnalazione per avvisare tutti i destinatari e i mittenti della fine dei lavori。

Esempio di codice:


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 + "终止了")
 
}

Condivisione della tecnologia