Stellen Sie go vor einen Funktionsaufruf, und diese Funktion läuft nebenläufig zum Rest des Programms, verwaltet von der Go-Runtime statt vom Betriebssystem. Eine zu starten ist der einfache Teil. Der schwierige Teil ist, zwei Goroutinen ein Ergebnis übergeben zu lassen, ohne dass sie sich gegenseitig den Zustand kaputt machen, und genau das ist die Aufgabe von Channels.

Was ist eine Goroutine?

Eine Goroutine ist eine Funktion, die unabhängig von der Funktion läuft, die sie gestartet hat, verwaltet von der Go-Runtime und nicht vom Betriebssystem. Ein OS-Thread reserviert ein Megabyte oder mehr Stack und kostet den Kernel echte Zeit für die Planung. Eine Goroutine startet mit einem Stack von etwa 2 KB, der bei Bedarf wächst, und die Runtime multiplext viele Goroutinen auf eine Handvoll OS-Threads. Tausende davon zu starten ist normal.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

package main

import (
	"fmt"
	"time"
)

func sayHi(name string) {
	fmt.Println("hi,", name)
}

func main() {
	go sayHi("alice")
	go sayHi("bob")
	time.Sleep(10 * time.Millisecond)
	fmt.Println("done")
}

Führen Sie das ein paar Mal aus, und die Reihenfolge von “hi, alice” und “hi, bob” ändert sich. Der Scheduler entscheidet. time.Sleep hält main nur lange genug am Leben, damit die Goroutinen drucken können: Entfernen Sie es, und das Programm kann beendet werden, bevor eine der beiden läuft, denn wenn main zurückkehrt, endet der Prozess, egal was sonst noch läuft. Der Sleep ist ein Workaround, keine Synchronisation. Ein Channel ist das, was ihn ersetzt.

Wie verbinden Channels Goroutinen?

Ein Channel ist eine typisierte Pipe: Eine Goroutine sendet einen Wert hinein, eine andere empfängt ihn, und die Runtime übernimmt die Übergabe. Stellen Sie sich ein einspuriges Förderband zwischen zwei Arbeitern vor. Der Sender legt ein Teil auf das Band und kann kein zweites auflegen, bevor das erste abgenommen wurde. Das ist ein ungepufferter Channel, und diese blockierende Übergabe ist der gesamte Synchronisationsmechanismus.

1
2
3
4
5
6
7
8

ch := make(chan string)

go func() {
	ch <- "payload ready"
}()

msg := <-ch
fmt.Println(msg)

make(chan string) erzeugt einen ungepufferten Channel für Strings. Das Senden (ch <- "payload ready") blockiert, bis etwas empfängt, und das Empfangen (<-ch) blockiert, bis etwas sendet. Keine Seite kommt weiter, bevor beide bereit sind, und das ist genau die Übergabe, die der time.Sleep oben nur vorgetäuscht hat.

Geben Sie dem Channel eine Kapazität, und das Band fasst mehr als ein Teil, bevor es sich verklemmt:

1
2
3
4
5

ch := make(chan int, 3)
ch <- 1
ch <- 2
ch <- 3
// a 4th send here would block until something receives

Das Senden eines gepufferten Channels blockiert nur, wenn der Puffer voll ist; das Empfangen blockiert nur, wenn er leer ist. Das lockert die Kopplung zwischen Sender und Empfänger, was hilft, Lastspitzen abzufedern. Es beseitigt nicht die Notwendigkeit der Koordination. Es verschiebt nur den Punkt, an dem das Blockieren beginnt.

Wie schließt man einen Channel und macht range darüber?

Der Sender schließt einen Channel, um zu sagen “es kommt nichts mehr”:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

func producer(ch chan<- int) {
	for i := 1; i <= 3; i++ {
		ch <- i
	}
	close(ch)
}

ch := make(chan int)
go producer(ch)
for v := range ch {
	fmt.Println(v) // 1, 2, 3, then the loop exits
}

for v := range ch empfängt weiter, bis der Channel geschlossen und leer ist, und endet dann von selbst. Kein Sentinel-Wert, keine manuelle Zählung. Um einen geschlossenen, leeren Channel von einem zu unterscheiden, der einen Nullwert geliefert hat, verwenden Sie die Zwei-Werte-Form: v, ok := <-ch setzt ok nur dann auf false, wenn der Channel geschlossen ist und nichts mehr übrig hat.

close hat drei Arten zu paniken, und der Compiler erkennt keine davon:

  • Senden auf einen geschlossenen Channel paniert.
  • Einen bereits geschlossenen Channel erneut schließen paniert.
  • Einen nil-Channel schließen paniert.

Deshalb schließt immer der Sender, niemals der Empfänger. Ein Empfänger hat keine Möglichkeit zu wissen, ob eine andere Goroutine noch senden will.

Wie wartet man, bis mehrere Goroutinen fertig sind?

range funktioniert, wenn ein Channel das Ende der Arbeit signalisiert. Wenn Sie mehrere Goroutinen gestartet haben und nur wissen müssen, dass alle fertig sind, verwenden Sie sync.WaitGroup:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

package main

import (
	"fmt"
	"sync"
)

func worker(id int, wg *sync.WaitGroup) {
	defer wg.Done()
	fmt.Printf("worker %d done\n", id)
}

func main() {
	var wg sync.WaitGroup
	for i := 1; i <= 3; i++ {
		wg.Add(1)
		go worker(i, &wg)
	}
	wg.Wait()
	fmt.Println("all workers finished")
}

Add(1) erhöht den Zähler, bevor jede Goroutine startet, Done() verringert ihn (per defer, läuft also auch bei einem Panic), und Wait() blockiert, bis der Zähler null erreicht. Verschieben Sie Add in die Goroutine, und Sie haben eine Race geschrieben: main kann Wait() erreichen, einen Zähler von null sehen und zurückkehren, bevor auch nur ein Worker ihn erhöht hat.

Wie funktioniert select mit mehreren Channels?

select wartet gleichzeitig auf mehrere Channel-Operationen und führt diejenige aus, die als Erste bereit ist. Timeouts und Abbruch in Go bauen darauf auf.

1
2
3
4
5
6
7
8

select {
case res := <-resultCh:
	fmt.Println("got result:", res)
case err := <-errCh:
	fmt.Println("failed:", err)
case <-time.After(2 * time.Second):
	fmt.Println("timed out waiting for a result")
}

Sind mehrere Cases bereit, wählt select eines davon zufällig aus, nicht in der geschriebenen Reihenfolge, die Reihenfolge der Cases gibt also keine Priorität. Ohne default und ohne bereiten Case blockiert select; ein leeres select{} blockiert für immer. Fügen Sie einen default-Case hinzu, und die gesamte Anweisung wird nicht blockierend: Ist in diesem Moment nichts bereit, läuft default.

Ein Worker-Pool kombiniert fast all das. Eine feste Anzahl von Goroutinen holt Jobs von einem Channel und schreibt Ergebnisse auf einen anderen, sodass die nebenläufige Arbeit begrenzt bleibt, egal wie viele Jobs ankommen.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

func worker(id int, jobs <-chan int, results chan<- int) {
	for j := range jobs {
		results <- j * j
	}
}

func main() {
	jobs := make(chan int, 100)
	results := make(chan int, 100)

	for w := 1; w <= 3; w++ {
		go worker(w, jobs, results)
	}

	for j := 1; j <= 9; j++ {
		jobs <- j
	}
	close(jobs)

	for i := 0; i < 9; i++ {
		fmt.Println(<-results)
	}
}

Drei Worker, neun Jobs. close(jobs) nach dem letzten Senden lässt die range-Schleife jedes Workers beenden, sobald die Warteschlange leer ist. Die Signaturen verwenden jobs <-chan int (nur empfangen) und results chan<- int (nur senden). Der Pfeil zeigt die Richtung, und der Compiler erzwingt sie, sodass ein Worker, der versehentlich auf jobs sendet, beim Kompilieren fehlschlägt statt sich zur Laufzeit falsch zu verhalten.

Warum leaken oder blockieren Goroutinen?

Goroutine-Leaks. Eine Goroutine, die für immer auf ein Senden oder Empfangen blockiert, das nie passieren wird, stürzt nicht ab. Sie bleibt dort, hält ihren Stack und alles, was sie per Closure erfasst hat, für die gesamte Lebensdauer des Programms. Die übliche Ursache ist eine Goroutine, die ein Ergebnis an einen Channel sendet, den niemand mehr liest, weil der Empfänger nach einem Timeout schon aufgegeben hat. Geben Sie der Goroutine einen Ausweg, meist einen context.Context, auf den sie ebenfalls per select wartet, damit sie das Senden abbrechen kann.

Deadlock. fatal error: all goroutines are asleep - deadlock! bedeutet, dass jede Goroutine, main eingeschlossen, auf einem Channel oder Lock blockiert ist, ohne dass noch etwas sie entblockieren könnte. Das kleinste Beispiel ist ein ungepuffertes Senden in main ohne eine andere laufende Goroutine, die es empfängt.

Races, die Channels nicht verhindern. Ein Channel synchronisiert die Übergabe eines Werts. Er schützt keine Variable, die Sie außerhalb des Channels weiter anfassen. Zwei Goroutinen, die denselben erfassten Zähler erhöhen, erzeugen trotzdem eine Race, Channel oder nicht, genau wie im Thread-Fall, der in was eine Race Condition ist und wie Sie sie beheben erklärt wird. Go liefert dafür einen Detector: Kompilieren oder testen Sie mit -race, und die Runtime meldet nebenläufige Zugriffe auf denselben Speicher, bei denen mindestens einer ein Schreibzugriff ist.

1
2

go test -race ./...
go build -race -o worker-pool .

Er erkennt nur Races, die während des Laufs tatsächlich auftreten, findet also unter realistischer nebenläufiger Last mehr als in einem ruhigen Unit-Test. Diese Last ist auch dort, wo goroutinenreiche Tests flaky werden, dieselbe Art von Unzuverlässigkeit, die in Unit-Tests vs Integrationstests vs End-to-End-Tests behandelt wird. Rechnen Sie mit einem Mehrfachen an Speicher und CPU bei aktivem -race. Führen Sie es in CI aus, nicht in der Binary, die Sie ausliefern.

Channel oder Mutex: was nehmen?

Beide machen nebenläufigen Zugriff sicher, aber von entgegengesetzten Enden aus. Ein sync.Mutex schützt gemeinsamen Zustand an Ort und Stelle: sperren, lesen oder schreiben, entsperren, und die Daten bewegen sich nie. Ein Channel verschiebt den Besitz eines Werts von einer Goroutine zu einer anderen, sodass konstruktionsbedingt immer nur eine Goroutine ihn anfasst. Gos eigene Empfehlung: Verwenden Sie Channels, um Arbeit zwischen Goroutinen zu koordinieren oder zu übergeben, und einen Mutex, um ein einzelnes gemeinsames Stück Daten zu schützen, das mehrere Goroutinen an Ort und Stelle aktualisieren, etwa einen Cache oder einen Zähler. Braucht eine Funktion beides, erledigt sie wahrscheinlich zwei Aufgaben, die sich getrennt besser lesen würden.

Wenn Sie von JavaScript kommen, ist das Modell anders. Ein Thread führt Ihren Code aus, und await gibt die Kontrolle nur an Stellen ab, die Sie im Quellcode sehen; Async/Await und die Event Loop in JavaScript erklärt zeigt, wo diese Suspendierungspunkte liegen. Go gibt Ihnen echte Parallelität, also liegt die Wahl des Primitivs bei Ihnen.

Für Ihren nächsten Batch-Job (einen Ordner mit Bildern verkleinern oder eine API einmal pro Zeile einer CSV-Datei aufrufen) starten Sie mit dem Worker-Pool oben. Wählen Sie die Worker-Anzahl, schließen Sie den Job-Channel, wenn Sie mit dem Senden fertig sind, und fügen Sie dem select einen context.Context hinzu, sobald Sie zum ersten Mal abbrechen müssen.

Verwandte Artikel