Golang Channels: gepuffert, ungepuffert, close und range

Senden und empfangen

Ein Channel ist eine typisierte Leitung zwischen Goroutinen. ch <- v sendet, <-ch empfängt. Du erzeugst einen mit make:

package main

import "fmt"

func main() {
	messages := make(chan string)

	go func() {
		messages <- "ping" // blocks until main receives
	}()

	msg := <-messages // blocks until the goroutine sends
	fmt.Println(msg)
}

Der Nullwert eines Channel-Typs ist nil, also liefert dir var ch chan string ohne make einen Channel, der für immer blockiert. Erzeuge Channels immer mit make.

Ungepufferte Channels synchronisieren

make(chan T) erzeugt einen ungepufferten Channel. Ein Senden blockiert, bis ein Empfänger den Wert nimmt, und ein Empfangen blockiert, bis ein Sender einen liefert. An diesem Punkt treffen sich die beiden Goroutinen, und das macht einen ungepufferten Channel ebenso zu einem Werkzeug der Synchronisation wie zu einer Datenleitung. Wenn <-done unten zurückkehrt, weißt du, dass der Worker alles vor seinem Senden erledigt hat:

package main

import "fmt"

func main() {
	done := make(chan struct{})
	result := 0

	go func() {
		for i := 1; i <= 100; i++ {
			result += i
		}
		done <- struct{}{} // signal: finished
	}()

	<-done
	fmt.Println("sum:", result)
}

result in main zu lesen ist hier ohne Mutex sicher. Das Speichermodell von Go garantiert, dass alles, was der Worker vor dem Senden getan hat, für main nach dem passenden Empfang sichtbar ist. chan struct{} ist der idiomatische Typ für ein reines Signal, weil struct{} keinen Speicher belegt.

Gepufferte Channels

make(chan T, n) gibt dem Channel Platz für n Werte. Senden gelingt ohne Empfänger, bis der Puffer voll ist; Empfangen gelingt, bis er leer ist. Die Werte kommen in der Reihenfolge heraus, in der sie hineingingen.

package main

import "fmt"

func main() {
	ch := make(chan int, 3)

	ch <- 10
	ch <- 20
	fmt.Println("len:", len(ch), "cap:", cap(ch))

	ch <- 30 // buffer now full; a fourth send here would block forever

	fmt.Println(<-ch, <-ch, <-ch)
	fmt.Println("len:", len(ch))
}

Ein Puffer entkoppelt Sender und Empfänger, damit kurze Spitzen den Sender nicht aufhalten. Einen Produzenten, der dauerhaft schneller ist als sein Konsument, repariert er nicht; er verschiebt nur den Moment, in dem der Sender blockiert. Wähl eine Puffergröße aus einem Grund (die Anzahl der Sender, eine bekannte Batch-Größe), nicht um einen Deadlock verschwinden zu lassen.

len(ch) ist eine Momentaufnahme. Bis du darauf reagierst, kann eine andere Goroutine den Wert schon geändert haben, also entscheide damit nicht, ob ein Senden blockieren wird. Dafür nimmst du select mit einem default-Case.

close und range

close(ch) sagt den Empfängern, dass keine Werte mehr gesendet werden. Nach einem close:

• werden Werte, die schon im Puffer liegen, weiterhin zugestellt,
• gibt danach jeder Empfang sofort den Nullwert zurück,
• meldet v, ok := <-ch ok == false,
• endet for v := range ch.

package main

import "fmt"

func produce(n int) <-chan int {
	ch := make(chan int)
	go func() {
		defer close(ch) // the sender closes, when it is done
		for i := range n {
			ch <- i * i
		}
	}()
	return ch
}

func main() {
	for v := range produce(5) {
		fmt.Println(v)
	}

	ch := make(chan string, 1)
	ch <- "last"
	close(ch)
	v, ok := <-ch
	fmt.Printf("%q %v\n", v, ok)
	v, ok = <-ch
	fmt.Printf("%q %v\n", v, ok)
}

Der erste Empfang nach close bekommt noch "last" mit ok == true. Der zweite bekommt den Nullwert "" und false.

Regeln, deren Verletzung eine Panic auslöst:

• Senden auf einen geschlossenen Channel löst eine Panic mit send on closed channel aus,
• einen bereits geschlossenen Channel zu schließen löst eine Panic aus,
• einen nil-Channel zu schließen löst eine Panic aus.

Also schließt nur die sendende Seite, und das nur einmal. Bei mehreren Sendern weiß kein einzelner, wann die anderen fertig sind; lass eine separate Goroutine auf alle Sender warten (mit einer sync.WaitGroup) und den Channel schließen, nachdem Wait zurückgekehrt ist. Schließen ist nur nötig, wenn ein Empfänger auf das Ende wartet. Ein nicht geschlossener Channel, auf den nichts mehr verweist, wird wie jeder andere Wert vom Garbage Collector eingesammelt.

Richtungstypen

Eine Funktion kann deklarieren, dass sie auf einem Channel nur sendet oder nur empfängt. Der Compiler lehnt dann die andere Operation ab.

Ein chan T wird implizit in jeden der eingeschränkten Typen konvertiert, wenn du ihn an eine Funktion übergibst. Deshalb kann produce oben einen <-chan int zurückgeben: Aufrufer können mit range darüber laufen, aber nicht hineinsenden oder ihn schließen. Nutz Richtungstypen bei jedem Funktionsparameter, wo sie passen. Sie dokumentieren die Zuständigkeit und machen Missbrauch zu einem Compilerfehler.

Deadlock

Wenn jede Goroutine blockiert ist und nichts eine davon wecken kann, bricht die Runtime das Programm ab:

package main

import "fmt"

func main() {
	ch := make(chan int)
	ch <- 1 // no other goroutine exists to receive
	fmt.Println(<-ch)
}

fatal error: all goroutines are asleep - deadlock!

goroutine 1 [chan send]:
main.main()
	/tmp/main.go:7 +0x38
exit status 2

Der Goroutine-Dump zeigt dir, welche Operation hängt (hier chan send, sonst etwa chan receive, sync.WaitGroup.Wait, select). Die üblichen Ursachen:

• ein Senden auf einen ungepufferten Channel, ohne dass ein Empfänger läuft,
• range über einen Channel, der nie geschlossen wird,
• ein WaitGroup-Zähler, der nie null erreicht,
• zwei Goroutinen, die jeweils auf die andere warten.

Die Runtime erkennt nur den Fall, in dem alle Goroutinen hängen. In einem Server mit anderen lebenden Goroutinen (ein HTTP-Listener, ein Ticker) bringt derselbe Bug nichts zum Absturz; die blockierte Goroutine leakt einfach.

nil-Channels

Senden und Empfangen auf einem nil-Channel blockieren für immer. Das klingt nutzlos, ist aber ein Standardtrick in select: Eine Channel-Variable auf nil zu setzen schaltet ihren Case ab. Dieses Beispiel führt zwei Channels zusammen und hört auf, auf einen zu lauschen, sobald er geschlossen ist:

package main

import "fmt"

func main() {
	a := make(chan int)
	b := make(chan int)
	go func() { defer close(a); a <- 1; a <- 2 }()
	go func() { defer close(b); b <- 10 }()

	sum := 0
	for a != nil || b != nil {
		select {
		case v, ok := <-a:
			if !ok {
				a = nil // a is finished: never select it again
				continue
			}
			sum += v
		case v, ok := <-b:
			if !ok {
				b = nil
				continue
			}
			sum += v
		}
	}
	fmt.Println("sum:", sum)
}

Ohne die Zuweisungen von nil ist ein geschlossener Channel immer bereit, und die Schleife würde auf Nullwerten durchdrehen.

Eine Pipeline

Channels lassen sich zu Pipelines verbinden: Jede Stufe ist eine Goroutine, die aus einem Channel empfängt, in den nächsten sendet und ihren Ausgang schließt, wenn ihr Eingang fertig ist.

package main

import "fmt"

func generate(nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for _, n := range nums {
			out <- n
		}
	}()
	return out
}

func square(in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			out <- n * n
		}
	}()
	return out
}

func main() {
	total := 0
	for v := range square(square(generate(1, 2, 3))) {
		fmt.Println(v)
		total += v
	}
	fmt.Println("total:", total)
}

Jede Stufe läuft nebenläufig, und die Ausgabereihenfolge ist deterministisch (1, 16, 81), weil jede Stufe eine einzelne Goroutine ist, die die Reihenfolge erhält. Das Schließen pflanzt sich fort: generate schließt, das beendet die range-Schleife von square, die schließt ihren Ausgang, und so weiter bis zu main.

Die Schwachstelle dieser Pipeline: Würde main früher aufhören zu lesen, würden die Stufen für immer beim Senden blockieren. Echte Pipelines nehmen einen context.Context oder einen done-Channel und warten mit select neben jedem Senden darauf.

Channel oder Mutex

Channels sind dafür da, die Zuständigkeit für Daten weiterzugeben und Ereignisse zu signalisieren. Ein sync.Mutex ist einfacher, um geteilten Zustand zu schützen, den viele Goroutinen direkt lesen und ändern, etwa einen Cache oder einen Zähler. Ein Struct mit einem Mutex darin ist oft klarer als eine Goroutine, die den Zustand besitzt und Anfragen über Channels bedient. Nimm, was den Code kürzer und die Zuständigkeit offensichtlich macht.

Kurzreferenz