Ολοκληρωμένα θέματα εφαρμογής των Go coroutines και channels

Ολοκληρωμένα θέματα εφαρμογής των κορουτινών και καναλιών Go

2024-07-12

1. Κατανοήστε εν συντομία τι είναι οι κορουτίνες και τα κανάλια

Τι είναι η κορουτίνα

Η κορουτίνα είναι ένα ελαφρύ νήμα σε επίπεδο χρήστη που έχει έναν ανεξάρτητο χώρο στοίβας και μοιράζεται τον χώρο σωρού του προγράμματος.

Είναι ένα micro-thread που υλοποιείται μέσω αλγορίθμων με βάση το single thread Σε σύγκριση με τον προγραμματισμό πολλαπλών νημάτων, έχει τα ακόλουθα πλεονεκτήματα:

Η εναλλαγή περιβάλλοντος της κορουτίνας αποφασίζεται από τον χρήστη, χωρίς την ανάγκη εναλλαγής περιβάλλοντος του πυρήνα του συστήματος, μειώνοντας τα γενικά έξοδα.
Από προεπιλογή, οι κορουτίνες προστατεύονται πλήρως για την αποφυγή διακοπών.Δεν απαιτείται κλείδωμα ατομικής λειτουργίας
Ένα μεμονωμένο νήμα μπορεί επίσης να επιτύχει υψηλή συγχρονικότητα, ενώ ακόμη και μια CPU μονού πυρήνα μπορεί να υποστηρίξει δεκάδες χιλιάδες κορουτίνες.

τι είναι κανάλι

Το κανάλι είναι μια δομή δεδομένων που χρησιμοποιείται για την επικοινωνία μεταξύ των κορουτινών. Παρόμοια με μια ουρά, το ένα άκρο είναι ο αποστολέας και το άλλο άκρο είναι ο παραλήπτης. Η χρήση καναλιών μπορεί να εξασφαλίσει τον συγχρονισμό και τη σειρά των δεδομένων.

Τα κανάλια χωρίζονται σε κανάλια buffer και κανάλια χωρίς buffer, τα οποία δηλώνονται ως εξής:

Υπάρχει ένα buffer κανάλι

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

κανάλι χωρίς προσωρινή αποθήκευση

intChan := make(chan int)

Η διαφορά μεταξύ καναλιών προσωρινής αποθήκευσης και καναλιών χωρίς προσωρινή αποθήκευση:

Αποκλεισμός: Ο αποστολέας ενός καναλιού χωρίς προσωρινή αποθήκευση θα μπλοκάρει έως ότου ληφθούν τα δεδομένα ο αποστολέας ενός καναλιού προσωρινής αποθήκευσης θα μπλοκάρει μέχρι να γεμίσει η προσωρινή μνήμη και ο δέκτης θα μπλοκάρει έως ότου η προσωρινή μνήμη δεν είναι άδεια.
Συγχρονισμός και αλληλουχία δεδομένων: Τα κανάλια χωρίς προσωρινή αποθήκευση εγγυώνται το συγχρονισμό και την ακολουθία δεδομένων.
Σενάρια εφαρμογών: Τα κανάλια χωρίς προσωρινή αποθήκευση απαιτούν αυστηρό συγχρονισμό και τα κανάλια με προσωρινή μνήμη μπορούν να επικοινωνούν ασύγχρονα και να βελτιώσουν την απόδοση.

Αυτό που πρέπει να σημειωθεί στην υλοποίηση καναλιών χωρίς προσωρινή αποθήκευση είναι ότι πρέπει να υπάρχει αποστολέας και παραλήπτης και στα δύο άκρα του καναλιού, διαφορετικά θα προκύψει αδιέξοδο.

2.Coroutine-channel concurrent programming case

(1) Εκτυπώστε γράμματα και αριθμούς εναλλάξ

Σημασία ερώτησης: Χρησιμοποιήστε το κανάλι κορουτίνας για να εκτυπώσετε εναλλάξ τους αριθμούς 1-10 και τα γράμματα AJ.

Κώδικας:


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

Ανάλυση θεμάτων:

Η ερώτηση απαιτεί να εκτυπώνουμε εναλλάξ γράμματα και αριθμούς, επομένως πρέπει να διασφαλίσουμε την αυστηρή σειρά των δύο κορουτινών, η οποία είναι συνεπής με το σενάριο εφαρμογής των καναλιών χωρίς προσωρινή αποθήκευση. Ρυθμίστε δύο κανάλια για αποθήκευση αριθμών και γραμμάτων αντίστοιχα. Δύο κορουτίνες που εκτυπώνουν αριθμούς και γράμματα χρησιμεύουν ως αποστολέας και δέκτης των δύο καναλιών αντίστοιχα. Εκτυπώστε μία φορά σε βρόχο και στείλτε ένα σήμα μία φορά για να υπενθυμίσετε σε άλλη κορουτίνα να εκτυπώσει.

Πρέπει να σημειωθεί ότι όταν εκτυπωθεί ο τελευταίος χαρακτήρας '10', η κορουτίνα για την εκτύπωση γραμμάτων έχει τελειώσει και το κανάλι numCh δεν έχει δέκτη Αυτή τη στιγμή, οι προϋποθέσεις υλοποίησης του καναλιού χωρίς προσωρινή μνήμη δεν πληρούνται - πρέπει να υπάρχουν ένας αποστολέας και ένας δέκτης Διαφορετικά, η αποστολή του σήματος ξανά θα προκαλέσει αδιέξοδο. Άρα δεν χρειάζεται να στείλετε ξανά το σήμα για 10η φορά.

(2) Σχεδιάστε έναν προγραμματιστή εργασιών

Τίτλος: Σχεδιάστε έναν χρονοπρογραμματιστή εργασιών που χρησιμοποιεί το μοντέλο προγραμματισμού πολυκορουτίνων + καναλιών για την υλοποίηση επιχειρηματικών σεναρίων ταυτόχρονης επεξεργασίας πολλαπλών εργασιών και απαιτεί η σειρά προγραμματισμού να είναι με τη σειρά με την οποία προστίθενται οι εργασίες.

Κώδικας:


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

ανάλυση προβλήματος:

Δεδομένου ότι οι προστιθέμενες εργασίες είναι πολλαπλές εργασίες, υπάρχουν περισσότερες από μία και απαιτείται ασύγχρονη επεξεργασία για την εκτέλεση αυτών των εργασιών. Η συμμόρφωση με κανάλια προσωρινής αποθήκευσης απαιτεί βελτιωμένη απόδοση και ασύγχρονη επεξεργασία.

Στη συνέχεια, πρέπει να βάλουμε την εργασία στο κανάλι και πολλοί δέκτες μπορούν να πάρουν τις εργασίες από το κανάλι με τη σειρά και να τις εκτελέσουν.

Το πρόβλημα που χρειάζεται προσοχή είναι ότι εάν ο αριθμός των εργασιών που προστίθενται είναι μεγαλύτερος από την προσωρινή μνήμη του καναλιού, θα προκαλέσει αποκλεισμό κατά την προσθήκη εργασιών. Για να μην επηρεαστεί η κανονική εκκίνηση του καταναλωτή, πρέπει να ανοίξει μια ξεχωριστή κορουτίνα για να προσθέσει εργασίες.

Με αυτόν τον τρόπο, όταν ο καταναλωτής καταναλώνει, ο παραγωγός αποκλεισμού θα αφυπνιστεί για να συνεχίσει να προσθέτει εργασίες.

3. Περίληψη

Αφού μελετήσαμε το μοντέλο προγραμματισμού κορουτίνας + καναλιών, εκτός από αυτό που μόλις αναφέρθηκε στον τίτλο, θα πρέπει να δώσουμε προσοχή και στα ακόλουθα ζητήματα:

1. Γιατί να κλείσει το κανάλι μετά την εξάντλησή του Ποιοι είναι οι κίνδυνοι να μην κλείσει;

Για αποφυγή αδιεξόδου. Το κλείσιμο του καναλιού ενημερώνει επίσης τον παραλήπτη ότι δεν υπάρχουν άλλα δεδομένα για αποστολή από τον αποστολέα και ότι δεν χρειάζεται να συνεχίσει να περιμένει δεδομένα. Μετά τη λήψη των πληροφοριών κλεισίματος του καναλιού, ο δέκτης σταματά να λαμβάνει δεδομένα εάν το κανάλι δεν είναι κλειστό, ο δέκτης θα παραμείνει αποκλεισμένος και υπάρχει κίνδυνος αδιεξόδου.
Αποδεσμεύστε πόρους και αποφύγετε τις διαρροές πόρων. Μετά το κλείσιμο του καναλιού, το σύστημα θα απελευθερώσει τους αντίστοιχους πόρους Το έγκαιρο κλείσιμο του καναλιού μπορεί να αποφύγει τη σπατάλη πόρων και τη διαρροή.

2. Πώς να κλείσετε το κανάλι με χάρη;

Καταρχάς, η πιο βασική αρχή του κλεισίματος των καναλιών είναι να μην κλείνουν κανάλια που έχουν κλείσει. Δεύτερον, υπάρχει μια άλλη αρχή για τη χρήση καναλιών Go:Μην κλείνετε το κανάλι στον δέκτη δεδομένων ή όταν υπάρχουν πολλοί αποστολείς.με άλλα λόγια，Θα πρέπει να επιτρέψουμε μόνο στον μοναδικό αποστολέα ενός καναλιού να κλείσει αυτό το κανάλι.

Ένας αγενής τρόπος είναι να κλείσετε το κανάλι μέσω ανάκτησης εξαιρέσεων, αλλά αυτό προφανώς παραβιάζει τις παραπάνω αρχές και μπορεί να προκαλέσει αγώνα δεδομένων. Ένας άλλος τρόπος είναι να κλείσετε το κανάλι με το sync.Once οι λειτουργίες και οι λειτουργίες αποστολής σε ένα κανάλι δεν δημιουργούν αγώνες δεδομένων. Και οι δύο μέθοδοι έχουν ορισμένα προβλήματα, επομένως δεν θα τις παρουσιάσω λεπτομερώς Εδώ είναι μια μέθοδος για το πώς να κλείσετε με χάρη το κανάλι.

Σενάριο 1: Μ δέκτες και ένας αποστολέας

Μια από τις πιο εύκολες καταστάσεις που αντιμετωπίζεις. Όταν ο αποστολέας πρέπει να ολοκληρώσει την αποστολή, απλώς αφήστε τον να κλείσει το κανάλι. Αυτό συμβαίνει στα δύο παραπάνω παραδείγματα προγραμματισμού.

Σενάριο 2: Ένας δέκτης και Ν αποστολείς

Σύμφωνα με τις βασικές αρχές των καναλιών Go, μπορούμε να κλείσουμε το κανάλι μόνο στον μοναδικό αποστολέα του καναλιού. Άρα, σε αυτή την περίπτωση, δεν μπορούμε να κλείσουμε κάπου απευθείας το κανάλι.Αλλά μπορούμε να αφήσουμε τον δέκτη να κλείσει ένα πρόσθετο κανάλι σήματος για να πει στον αποστολέα να μην στείλει άλλα δεδομένα.。


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

Σε αυτή τη μέθοδο, προσθέτουμε ένα πρόσθετο κανάλι σήματος stopCh, το οποίο χρησιμοποιεί ο δέκτης για να πει στον αποστολέα ότι δεν χρειάζεται πλέον να λαμβάνει δεδομένα. Επιπλέον, αυτή η μέθοδος δεν κλείνει dataCh Όταν ένα κανάλι δεν χρησιμοποιείται πλέον από καμία κορουτίνα, θα συλλέγεται σταδιακά σκουπίδια, ανεξάρτητα από το αν έχει κλείσει.

Η κομψότητα αυτής της μεθόδου είναι ότι κλείνοντας ένα κανάλι, σταματάτε να χρησιμοποιείτε άλλο κανάλι, κλείνοντας έτσι έμμεσα το άλλο κανάλι.

Σενάριο 3: M δέκτες και N αποστολείς

Δεν μπορούμε ούτε ο δέκτης ούτε ο αποστολέας να κλείσει το κανάλι που χρησιμοποιείται για τη μετάδοση των δεδομένων, ούτε μπορούμε να βάλουμε έναν από τους πολλαπλούς δέκτες να κλείσει ένα πρόσθετο κανάλι σηματοδότησης. Και οι δύο αυτές πρακτικές παραβιάζουν την αρχή του κλεισίματος του καναλιού.

Ωστόσο, μπορούμε να εισαγάγουμεΈνας ρόλος ενδιάμεσου διαμεσολαβητή και να κλείνει πρόσθετα κανάλια σηματοδότησης για να ειδοποιεί όλους τους δέκτες και τους αποστολείς για το τέλος της εργασίας。

Παράδειγμα κώδικα:


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

Κοινή χρήση τεχνολογίας