--- id: 01933e13f963942e4c9bc8c3c89fbc2b author: snowmerak title: Go-Starterpaket für Nebenläufigkeit description: Wir stellen verschiedene Werkzeuge und Techniken für die Verwaltung der Nebenläufigkeit in Go vor, darunter Goroutinen, Kanäle und Mutexe. language: de date: 2024-10-18T05:51:30.010680581Z path: /blog/posts/go-concurrency-starter-pack-z221a3399 category: concurrency --- ## Überblick ### Kurze Einführung Die Sprache Go verfügt über viele Werkzeuge zur Verwaltung von Nebenläufigkeit. In diesem Artikel möchte ich einige davon sowie einige Tricks vorstellen. ### Goroutinen? Eine goroutine ist ein neuartiges Nebenläufigkeitsmodell, das von der Sprache Go unterstützt wird. Üblicherweise erhält ein Programm vom Betriebssystem OS-Threads, um mehrere Aufgaben gleichzeitig auszuführen, und arbeitet dabei parallel entsprechend der Anzahl der Kerne. Um Nebenläufigkeit auf einer feineren Ebene zu realisieren, erzeugt man im Userland grüne Threads, sodass innerhalb eines einzelnen OS-Threads mehrere grüne Threads abwechselnd Aufgaben ausführen. Goroutinen wurden jedoch als eine Form solcher grünen Threads entwickelt, die noch kleiner und effizienter sind. Goroutinen benötigen weniger Speicher als Threads und können schneller erzeugt und umgeschaltet werden. Um Goroutinen zu verwenden, genügt es, einfach das Schlüsselwort `go` zu benutzen. Dadurch lässt sich bei der Programmierung intuitiv synchroner Code als asynchroner Code ausführen. ```go package main import ( "fmt" "time" ) func main() { ch := make(chan struct{}) go func() { defer close(ch) time.Sleep(1 * time.Second) fmt.Println("Hello, World!") }() fmt.Println("Waiting for goroutine...") for range ch {} } ``` Dieser Code wandelt synchronen Code, der eine Sekunde wartet und anschließend `Hello, World!` ausgibt, in einen asynchronen Ablauf um. Das gegenwärtige Beispiel ist einfach, doch wenn man etwas komplexeren Code von synchron nach asynchron überführt, verbessern sich Lesbarkeit, Übersichtlichkeit und Verständlichkeit gegenüber Ansätzen wie async/await oder Promises. In vielen Fällen entsteht jedoch schlechter Goroutinen-Code, wenn man den Ablauf, solchen synchronen Code einfach asynchron aufzurufen, sowie Abläufe wie `fork & join` (ähnlich wie bei Divide and Conquer) nicht versteht. Um solchen Fällen zu begegnen, stelle ich in diesem Artikel einige Methoden und Techniken vor. ## Verwaltung von Nebenläufigkeit ### context Dass `context` als erstes Verwaltungswerkzeug erscheint, mag überraschen. In Go übernimmt `context` jedoch über die einfache Abbruchfunktion hinaus hervorragend die Verwaltung des gesamten Aufgabenbaums. Für diejenigen, die das Paket nicht kennen, erkläre ich es kurz. ```go package main func main() { ctx, cancel := context.WithCancel(context.Background()) defer cancel() go func() { <-ctx.Done() fmt.Println("Context is done!") }() time.Sleep(1 * time.Second) cancel() time.Sleep(1 * time.Second) } ``` Der obige Code verwendet `context`, um nach einer Sekunde `Context is done!` auszugeben. Über die Methode `Done()` lässt sich prüfen, ob abgebrochen wurde, und mit Methoden wie `WithCancel`, `WithTimeout`, `WithDeadline` und `WithValue` stehen verschiedene Abbruchmethoden zur Verfügung. Lassen Sie uns ein einfaches Beispiel erstellen. Angenommen, Sie möchten mit dem Muster `aggregator` die Daten `user`, `post` und `comment` abrufen, und alle Anfragen müssen innerhalb von zwei Sekunden erfolgen. Dann können Sie wie folgt schreiben. ```go package main func main() { ctx, cancel := context.WithTimeout(context.Background(), 2 * time.Second) defer cancel() ch := make(chan struct{}) go func() { defer close(ch) user := getUser(ctx) post := getPost(ctx) comment := getComment(ctx) fmt.Println(user, post, comment) }() select { case <-ctx.Done(): fmt.Println("Timeout!") case <-ch: fmt.Println("All data is fetched!") } } ``` Der obige Code gibt `Timeout!` aus, wenn nicht alle Daten innerhalb von zwei Sekunden abgerufen werden konnten, und `All data is fetched!`, wenn alle Daten abgerufen wurden. Auf diese Weise lassen sich Abbruch und Timeout auch in Code mit mehreren laufenden Goroutinen leicht verwalten, wenn man `context` verwendet. Weitere Funktionen und Methoden im Zusammenhang mit context finden Sie in der [godoc context](https://pkg.go.dev/context). Ich hoffe, dass Sie das Einfache erlernen und bequem nutzen können. ### channel #### unbuffered channel Ein `channel` ist ein Werkzeug zur Kommunikation zwischen Goroutinen. Ein `channel` kann mit `make(chan T)` erzeugt werden. Dabei ist `T` der Typ der Daten, die der jeweilige `channel` übertragen soll. Mit `<-` können Daten gesendet und empfangen werden, und mit `close` lässt sich ein `channel` schließen. ```go package main func main() { ch := make(chan int) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` Der obige Code gibt mithilfe eines `channel` 1 und 2 aus. In diesem Code wird lediglich gezeigt, wie Werte an einen `channel` gesendet und von ihm empfangen werden. Ein `channel` bietet jedoch weit mehr Funktionen. Zunächst betrachten wir `buffered channel` und `unbuffered channel`. Vorab sei gesagt, dass das oben gezeigte Beispiel ein `unbuffered channel` ist, bei dem das Senden und das Empfangen von Daten gleichzeitig erfolgen müssen. Wenn diese Vorgänge nicht gleichzeitig stattfinden, kann ein Deadlock entstehen. #### buffered channel Wie wäre es, wenn der obige Code keine einfache Ausgabe, sondern zwei Prozesse mit aufwendigen Aufgaben wäre? Wenn der zweite Prozess beim Lesen und Verarbeiten längere Zeit blockiert, würde auch der erste Prozess in dieser Zeit angehalten. Um diese Situation zu vermeiden, können wir einen `buffered channel` verwenden. ```go package main func main() { ch := make(chan int, 2) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` Der obige Code gibt mithilfe eines `buffered channel` 1 und 2 aus. In diesem Code wird ein `buffered channel` verwendet, sodass das Senden und das Empfangen von Daten nicht gleichzeitig erfolgen müssen. Durch einen Puffer im Channel entsteht Spielraum in der entsprechenden Länge, wodurch sich Verzögerungen vermeiden lassen, die durch die Last nachrangiger Aufgaben entstehen. #### select Wenn Sie mit mehreren Kanälen arbeiten, lässt sich mit der Anweisung `select` leicht eine `fan-in`-Struktur implementieren. ```go package main import ( "fmt" "time" ) func main() { ch1 := make(chan int, 10) ch2 := make(chan int, 10) ch3 := make(chan int, 10) go func() { for { ch1 <- 1 time.Sleep(1 * time.Second) } }() go func() { for { ch2 <- 2 time.Sleep(2 * time.Second) } }() go func() { for { ch3 <- 3 time.Sleep(3 * time.Second) } }() for i := 0; i < 3; i++ { select { case v := <-ch1: fmt.Println(v) case v := <-ch2: fmt.Println(v) case v := <-ch3: fmt.Println(v) } } } ``` Der obige Code erzeugt drei Kanäle, die periodisch 1, 2 und 3 übermitteln, und gibt die über `select` aus den Kanälen empfangenen Werte aus. Auf diese Weise können Sie mit `select` gleichzeitig Daten aus mehreren Kanälen empfangen und jeden Wert verarbeiten, sobald er eintrifft. #### for range Ein `channel` lässt sich mit `for range` leicht auslesen. Wird `for range` auf einen Kanal angewendet, wird es bei jedem Hinzufügen von Daten zu diesem Kanal ausgeführt, und die Schleife endet, sobald der Kanal geschlossen wird. ```go package main func main() { ch := make(chan int) go func() { ch <- 1 ch <- 2 close(ch) }() for i := range ch { fmt.Println(i) } } ``` Der obige Code gibt mithilfe eines `channel` 1 und 2 aus. In diesem Code werden mit `for range` bei jedem Hinzufügen von Daten zum Kanal die Daten empfangen und ausgegeben. Wird der Kanal geschlossen, endet die Schleife. Wie bereits mehrfach gezeigt, kann diese Syntax auch als einfaches Synchronisationsmittel verwendet werden. ```go package main func main() { ch := make(chan struct{}) go func() { defer close(ch) time.Sleep(1 * time.Second) fmt.Println("Hello, World!") }() fmt.Println("Waiting for goroutine...") for range ch {} } ``` Der obige Code wartet eine Sekunde und gibt danach `Hello, World!` aus. In diesem Code wird ein `channel` verwendet, um synchronen Code in asynchronen Code umzuwandeln. Auf diese Weise lässt sich mit einem `channel` synchroner Code leicht in asynchronen Code überführen und ein `join`-Punkt festlegen. #### etc 1. Wenn man Daten an einen nil channel sendet oder von ihm empfängt, kann es zu einer Endlosschleife und somit zu einem Deadlock kommen. 2. Wenn man nach dem Schließen eines Kanals Daten sendet, tritt ein panic auf. 3. Auch wenn man einen Kanal nicht ausdrücklich schließt, schließt ihn der GC beim Einsammeln. ### mutex #### spinlock Ein `spinlock` ist ein Synchronisationsverfahren, das in einer Schleife wiederholt versucht, die Sperre zu erlangen. In Go lässt sich mithilfe von Zeigern leicht ein Spinlock implementieren. ```go package spinlock import ( "runtime" "sync/atomic" ) type SpinLock struct { lock uintptr } func (s *SpinLock) Lock() { for !atomic.CompareAndSwapUintptr(&s.lock, 0, 1) { runtime.Gosched() } } func (s *SpinLock) Unlock() { atomic.StoreUintptr(&s.lock, 0) } func NewSpinLock() *SpinLock { return &SpinLock{} } ``` Der obige Code ist eine Implementierung des Pakets `spinlock`. Hier wird `SpinLock` mithilfe des Pakets `sync/atomic` realisiert. In der Methode `Lock` wird mit `atomic.CompareAndSwapUintptr` versucht, die Sperre zu setzen, und in der Methode `Unlock` wird mit `atomic.StoreUintptr` die Sperre freigegeben. Da dieses Verfahren ununterbrochen versucht, die Sperre zu erlangen, verbraucht es bis zum Erhalt der Sperre fortlaufend CPU-Leistung und kann in eine Endlosschleife geraten. Daher ist es ratsam, `spinlock` nur für einfache Synchronisation oder für sehr kurze Zeiträume zu verwenden. #### sync.Mutex `mutex` ist ein Werkzeug zur Synchronisation zwischen Goroutinen. Das vom Paket `sync` bereitgestellte `mutex` bietet Methoden wie `Lock`, `Unlock`, `RLock` und `RUnlock`. Ein `mutex` kann als `sync.Mutex` erzeugt werden, und mit `sync.RWMutex` lässt sich außerdem eine Lese-/Schreibsperre verwenden. ```go package main import ( "sync" ) func main() { var mu sync.Mutex var count int go func() { mu.Lock() count++ mu.Unlock() }() mu.Lock() count++ mu.Unlock() println(count) } ``` Im obigen Code greifen zwei Goroutinen nahezu gleichzeitig auf dieselbe Variable `count` zu. Wenn man den Code, der auf die Variable `count` zugreift, mithilfe eines `mutex` zu einem kritischen Abschnitt macht, lässt sich der gleichzeitige Zugriff auf `count` verhindern. Dann gibt dieser Code unabhängig von der Anzahl der Ausführungen stets `2` aus. #### sync.RWMutex `sync.RWMutex` ist ein `mutex`, bei dem Lesesperren und Schreibsperren unterschieden werden. Mit den Methoden `RLock` und `RUnlock` können Sie eine Lesesperre setzen und wieder aufheben. ```go package cmap import ( "sync" ) type ConcurrentMap[K comparable, V any] struct { sync.RWMutex data map[K]V } func (m *ConcurrentMap[K, V]) Get(key K) (V, bool) { m.RLock() defer m.RUnlock() value, ok := m.data[key] return value, ok } func (m *ConcurrentMap[K, V]) Set(key K, value V) { m.Lock() defer m.Unlock() m.data[key] = value } ``` Der obige Code implementiert `ConcurrentMap` mithilfe von `sync.RWMutex`. In diesem Code wird in der Methode `Get` eine Lesesperre und in der Methode `Set` eine Schreibsperre gesetzt, sodass auf die Map `data` sicher zugegriffen und sie verändert werden kann. Die Lesesperre ist deshalb notwendig, weil bei überwiegend lesenden Operationen ohne Schreibsperre nur eine Lesesperre gesetzt wird, damit mehrere Goroutinen gleichzeitig lesen können. Dadurch lässt sich die Leistung verbessern, wenn keine Zustandsänderung erfolgt und somit keine Schreibsperre nötig ist. #### fakelock `fakelock` ist ein einfacher Trick, bei dem `sync.Locker` implementiert wird. Diese Struktur bietet dieselben Methoden wie `sync.Mutex`, führt jedoch tatsächlich keinerlei Aktion aus. ```go package fakelock type FakeLock struct{} func (f *FakeLock) Lock() {} func (f *FakeLock) Unlock() {} ``` Der obige Code ist eine Implementierung des Pakets `fakelock`. Dieses Paket implementiert `sync.Locker` und stellt die Methoden `Lock` und `Unlock` bereit, führt jedoch tatsächlich keinerlei Operation aus. Warum dieser Code benötigt wird, werde ich bei Gelegenheit erläutern. ### waitgroup #### sync.WaitGroup `sync.WaitGroup` ist ein Werkzeug, das wartet, bis alle Aufgaben von Goroutinen abgeschlossen sind. Es bietet die Methoden `Add`, `Done` und `Wait`. Mit der Methode `Add` wird die Anzahl der Goroutinen erhöht, mit `Done` wird signalisiert, dass die Aufgabe einer Goroutine beendet ist. Mit `Wait` wird gewartet, bis die Aufgaben aller Goroutinen abgeschlossen sind. ```go package main import ( "sync" "sync/atomic" ) func main() { wg := sync.WaitGroup{} c := atomic.Int64{} for i := 0; i < 100 ; i++ { wg.Add(1) go func() { defer wg.Done() c.Add(1) }() } wg.Wait() println(c.Load()) } ``` Der obige Code verwendet `sync.WaitGroup`, damit 100 Goroutinen gleichzeitig einen Wert zur Variable `c` addieren. In diesem Code wartet `sync.WaitGroup`, bis alle Goroutinen abgeschlossen sind, und anschließend wird der in `c` addierte Wert ausgegeben. Bei einem einfachen `fork & join` einiger weniger Aufgaben genügt es, nur Channels zu verwenden, während bei einem `fork & join` einer großen Anzahl von Aufgaben `sync.WaitGroup` eine gute Wahl sein kann. #### with slice In Verbindung mit einem Slice kann `waitgroup` ein hervorragendes Werkzeug zur Verwaltung nebenläufiger Ausführung ohne Sperren werden. ```go package main import ( "fmt" "sync" "rand" ) func main() { var wg sync.WaitGroup arr := [10]int{} for i := 0; i < 10; i++ { wg.Add(1) go func(id int) { defer wg.Done() arr[id] = rand.Intn(100) }(i) } wg.Wait() fmt.Println("Done") for i, v := range arr { fmt.Printf("arr[%d] = %d\n", i, v) } } ``` Der obige Code erzeugt mit `waitgroup` allein in jeder Goroutine gleichzeitig jeweils eine von 10 Zufallszahlen und speichert sie am zugewiesenen Index. In diesem Code wartet `waitgroup`, bis alle Goroutinen abgeschlossen sind, und gibt danach `Done` aus. Auf diese Weise können bei Verwendung von `waitgroup` mehrere Goroutinen gleichzeitig Aufgaben ausführen, Daten ohne Sperren speichern, bis alle Goroutinen beendet sind, und danach gesammelt nachbearbeitet werden. #### golang.org/x/sync/errgroup.ErrGroup `errgroup` ist ein Paket, das `sync.WaitGroup` erweitert. Im Gegensatz zu `sync.WaitGroup` bricht `errgroup` alle Goroutinen ab und gibt den Fehler zurück, sobald in einer der Aufgaben der Goroutinen ein Fehler auftritt. ```go package main import ( "context" "fmt" "golang.org/x/sync/errgroup" ) func main() { g, ctx := errgroup.WithContext(context.Background()) _ = ctx for i := 0; i < 10; i++ { i := i g.Go(func() error { if i == 5 { return fmt.Errorf("error") } return nil }) } if err := g.Wait(); err != nil { fmt.Println(err) } } ``` Der obige Code erzeugt mit `errgroup` 10 Goroutinen, wobei in der fünften Goroutine ein Fehler ausgelöst wird. Der Fehler in der fünften Goroutine wurde absichtlich ausgelöst, um den Fall eines Fehlers zu zeigen. In der tatsächlichen Anwendung erzeugt man Goroutinen mit `errgroup` und führt dann für den Fall, dass in einer Goroutine ein Fehler auftritt, verschiedene Nachbearbeitungen durch. ### once Ein Werkzeug, um Code auszuführen, der nur einmal ausgeführt werden darf. Über die folgenden Konstruktoren lässt sich der entsprechende Code ausführen. ```go func OnceFunc(f func()) func() func OnceValue[T any](f func() T) func() T func OnceValues[T1, T2 any](f func() (T1, T2)) func() (T1, T2) ``` #### OnceFunc `OnceFunc` sorgt einfach dafür, dass die jeweilige Funktion über die gesamte Laufzeit hinweg genau einmal ausgeführt werden kann. ```go package main import "sync" func main() { once := sync.OnceFunc(func() { println("Hello, World!") }) once() once() once() once() once() } ``` Der obige Code gibt mithilfe von `sync.OnceFunc` `Hello, World!` aus. In diesem Code wird mit `sync.OnceFunc` die Funktion `once` erzeugt, und obwohl `once` mehrfach aufgerufen wird, wird `Hello, World!` nur einmal ausgegeben. #### OnceValue `OnceValue` sorgt nicht einfach nur dafür, dass die jeweilige Funktion über die gesamte Laufzeit hinweg genau einmal ausgeführt wird, sondern speichert zusätzlich den Rückgabewert der Funktion und gibt bei erneuten Aufrufen den gespeicherten Wert zurück. ```go package main import "sync" func main() { c := 0 once := sync.OnceValue(func() int { c += 1 return c }) println(once()) println(once()) println(once()) println(once()) println(once()) } ``` Der obige Code erhöht mithilfe von `sync.OnceValue` die Variable `c` um jeweils 1. In diesem Code wird mit `sync.OnceValue` die Funktion `once` erzeugt, und obwohl `once` mehrfach aufgerufen wird, wird die Variable `c` nur einmal erhöht und liefert den Wert 1 zurück. #### OnceValues `OnceValues` funktioniert genauso wie `OnceValue`, kann jedoch mehrere Werte zurückgeben. ```go package main import "sync" func main() { c := 0 once := sync.OnceValues(func() (int, int) { c += 1 return c, c }) a, b := once() println(a, b) a, b = once() println(a, b) a, b = once() println(a, b) a, b = once() println(a, b) a, b = once() println(a, b) } ``` Der obige Code erhöht mithilfe von `sync.OnceValues` die Variable `c` um jeweils 1. In diesem Code wird mit `sync.OnceValues` die Funktion `once` erzeugt, und obwohl `once` mehrfach aufgerufen wird, wird die Variable `c` nur einmal erhöht und liefert den Wert 1 zurück. ### atomic Das Paket `atomic` stellt atomare Operationen bereit. Es bietet Methoden wie `Add`, `CompareAndSwap`, `Load`, `Store` und `Swap`, empfohlen wird jedoch in letzter Zeit die Verwendung von Typen wie `Int64`, `Uint64` und `Pointer`. ```go package main import ( "sync" "sync/atomic" ) func main() { wg := sync.WaitGroup{} c := atomic.Int64{} for i := 0; i < 100 ; i++ { wg.Add(1) go func() { defer wg.Done() c.Add(1) }() } wg.Wait() println(c.Load()) } ``` Dies ist das Beispiel, das wir zuvor verwendet haben. Es zeigt Code, der den Typ `atomic.Int64` verwendet, um die Variable `c` atomar zu erhöhen. Mit den Methoden `Add` und `Load` lässt sich die Variable atomar erhöhen und auslesen. Außerdem kann mit der Methode `Store` ein Wert gespeichert, mit der Methode `Swap` ein Wert ausgetauscht und mit der Methode `CompareAndSwap` ein Wert verglichen und bei Übereinstimmung ersetzt werden. ### cond #### sync.Cond Das Paket `cond` stellt Bedingungsvariablen bereit. Es lässt sich mit `sync.Cond` erzeugen und bietet die Methoden `Wait`, `Signal` und `Broadcast`. ```go package main import ( "sync" ) func main() { c := sync.NewCond(&sync.Mutex{}) ready := false go func() { c.L.Lock() ready = true c.Signal() c.L.Unlock() }() c.L.Lock() for !ready { c.Wait() } c.L.Unlock() println("Ready!") } ``` Der obige Code wartet mithilfe von `sync.Cond`, bis die Variable `ready` den Wert `true` annimmt. In diesem Code wird mit `sync.Cond` gewartet, bis die Variable `ready` `true` wird, und danach wird `Ready!` ausgegeben. Auf diese Weise lässt sich mit `sync.Cond` erreichen, dass mehrere Goroutinen gleichzeitig warten, bis eine bestimmte Bedingung erfüllt ist. Darauf aufbauend lässt sich eine einfache `queue` implementieren. ```go package queue import ( "sync" "sync/atomic" ) type Node[T any] struct { Value T Next *Node[T] } type Queue[T any] struct { sync.Mutex Cond *sync.Cond Head *Node[T] Tail *Node[T] Len int } func New[T any]() *Queue[T] { q := &Queue[T]{} q.Cond = sync.NewCond(&q.Mutex) return q } func (q *Queue[T]) Push(value T) { q.Lock() defer q.Unlock() node := &Node[T]{Value: value} if q.Len == 0 { q.Head = node q.Tail = node } else { q.Tail.Next = node q.Tail = node } q.Len++ q.Cond.Signal() } func (q *Queue[T]) Pop() T { q.Lock() defer q.Unlock() for q.Len == 0 { q.Cond.Wait() } node := q.Head q.Head = q.Head.Next q.Len-- return node.Value } ``` Wenn man `sync.Cond` auf diese Weise nutzt, kann man effizient warten, statt mit einem `spin-lock` viel CPU-Last zu erzeugen, und wieder weiterarbeiten, sobald die Bedingung erfüllt ist. ### semaphore #### golang.org/x/sync/semaphore.Semaphore Das Paket `semaphore` stellt Semaphoren bereit. Es lässt sich mit `golang.org/x/sync/semaphore.Semaphore` erzeugen und bietet die Methoden `Acquire`, `Release` und `TryAcquire`. ```go package main import ( "context" "fmt" "golang.org/x/sync/semaphore" ) func main() { s := semaphore.NewWeighted(1) if s.TryAcquire(1) { fmt.Println("Acquired!") } else { fmt.Println("Not Acquired!") } s.Release(1) } ``` Der obige Code erzeugt mit `semaphore` eine Semaphore und zeigt, wie sie mit der Methode `Acquire` erworben und mit der Methode `Release` wieder freigegeben wird. In diesem Code wird gezeigt, wie man mit `semaphore` eine Semaphore erwirbt und wieder freigibt. ## Abschließende Worte Die grundlegenden Inhalte dürften hiermit ausreichen. Ich hoffe, dass Sie auf Grundlage dieses Artikels verstehen, wie man Nebenläufigkeit mithilfe von Goroutinen verwaltet, und es praktisch einsetzen können. Ich hoffe, dass dieser Artikel für Sie hilfreich war. Vielen Dank.