Postaw go przed wywołaniem funkcji i ta funkcja zacznie działać współbieżnie z resztą programu, zarządzana przez runtime Go, a nie przez system operacyjny. Uruchomienie jej to łatwa część. Trudna część to przekazanie wyniku między dwoma goroutines bez wchodzenia sobie w drogę przy współdzielonym stanie, a tym właśnie zajmują się channels.

Co to jest goroutine?

Goroutine to funkcja działająca niezależnie od tej, która ją uruchomiła. Wątek systemu operacyjnego rezerwuje megabajt lub więcej stosu i kosztuje jądro realny czas na szeregowanie. Goroutine startuje ze stosem około 2KB, który rośnie w razie potrzeby, a runtime multiplexuje wiele goroutines na garstkę wątków systemowych. Uruchamianie ich tysięcy naraz jest normalne.

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

Uruchom to kilka razy i kolejność “hi, alice” oraz “hi, bob” się zmienia. Decyduje scheduler. time.Sleep tylko utrzymuje main przy życiu wystarczająco długo, by goroutines zdążyły wypisać tekst: usuń go i program może zakończyć się, zanim któraś z nich zadziała, bo powrót z main kończy proces niezależnie od tego, co jeszcze trwa. Sleep to obejście, nie synchronizacja. Channel jest tym, co go zastępuje.

Jak channels łączą goroutines?

Channel to typowana rura: jedna goroutine wysyła do niej wartość, druga ją odbiera, a runtime obsługuje przekazanie. Wyobraź sobie jednotorowy przenośnik między dwoma pracownikami. Nadawca kładzie jeden element na przenośniku i nie może położyć drugiego, dopóki pierwszy nie zostanie zdjęty. To jest channel bez bufora, i to blokujące przekazanie jest całym mechanizmem synchronizacji.

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) tworzy channel stringów bez bufora. Wysłanie (ch <- "payload ready") blokuje, dopóki coś go nie odbierze, a odebranie (<-ch) blokuje, dopóki coś go nie wyśle. Żadna ze stron nie rusza dalej, aż obie są gotowe, i to jest właśnie przekazanie, które time.Sleep wcześniej tylko udawał.

Daj channel pewną pojemność i przenośnik pomieści więcej niż jeden element, zanim się zatka:

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

Wysłanie do channel z buforem blokuje tylko wtedy, gdy bufor jest pełny; odebranie blokuje tylko wtedy, gdy jest pusty. To rozluźnia zależność między nadawcą a odbiorcą, co pomaga wchłonąć skoki ruchu. Nie usuwa potrzeby koordynacji. Przesuwa tylko moment, w którym zaczyna się blokowanie.

Jak zamknąć channel i zrobić po nim range?

Nadawca zamyka channel, by powiedzieć “nic więcej nie przyjdzie”:

 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 odbiera dalej, aż channel zostanie zamknięty i opróżniony, a potem sam się kończy. Żadnej wartości wartownika, żadnego ręcznego liczenia. Aby odróżnić zamknięty, pusty channel od takiego, który dostarczył wartość zerową, użyj formy dwuwartościowej: v, ok := <-ch ustawia ok na false tylko wtedy, gdy channel jest zamknięty i nic już nie zostało.

close ma trzy sposoby na panikę, i kompilator nie wychwytuje żadnego z nich:

  • Wysłanie do zamkniętego channel panikuje.
  • Zamknięcie już zamkniętego channel panikuje.
  • Zamknięcie channel nil panikuje.

Dlatego zamyka zawsze nadawca, nigdy odbiorca. Odbiorca nie ma sposobu, by wiedzieć, czy inna goroutine wciąż zamierza wysłać coś jeszcze.

Jak czekać na zakończenie wielu goroutines?

range działa, gdy channel sygnalizuje koniec pracy. Gdy uruchomiłeś kilka goroutines i potrzebujesz tylko wiedzieć, że wszystkie się zakończyły, użyj 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) zwiększa licznik przed startem każdej goroutine, Done() go zmniejsza (przez defer, więc działa nawet po panice funkcji), a Wait() blokuje, aż licznik dojdzie do zera. Przenieś Add do wnętrza goroutine i masz napisaną race: main może dojść do Wait(), zobaczyć licznik na zero i wrócić, zanim choć jeden worker go zwiększył.

Jak select działa z wieloma channels?

select czeka jednocześnie na kilka operacji na channels i wykonuje tę, która jest gotowa pierwsza. Timeouty i anulowanie w Go są zbudowane na tym.

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

Jeśli gotowy jest więcej niż jeden case, select wybiera jeden losowo, nie w kolejności zapisu, więc kolejność case’ów nie daje żadnego priorytetu. Bez default i bez gotowego case’a select blokuje; puste select{} blokuje na zawsze. Dodaj case default, a cała instrukcja staje się nieblokująca: jeśli w tym momencie nic nie jest gotowe, wykonuje się default.

Worker pool łączy w sobie prawie wszystko z tego. Ustalona liczba goroutines pobiera jobs z jednego channel i wypycha wyniki na drugi, dzięki czemu współbieżna praca ma stały limit niezależnie od liczby nadchodzących jobs.

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

Trzy workers, dziewięć jobs. close(jobs) po ostatnim wysłaniu pozwala, by każdy worker zakończył swoją pętlę range, gdy kolejka się opróżni. Sygnatury używają jobs <-chan int (tylko odbiór) i results chan<- int (tylko wysyłanie). Strzałka wskazuje kierunek, a kompilator go wymusza, więc worker, który przez błąd wysyła do jobs, nie skompiluje się, zamiast źle działać w runtime.

Dlaczego goroutines leakują lub wchodzą w deadlock?

Leaki goroutines. Goroutine zablokowana na zawsze na wysłaniu lub odebraniu, które nigdy nie nastąpi, nie crashuje. Po prostu tam siedzi, trzymając swój stos i wszystko, co przechwyciła przez closure, przez cały czas życia programu. Zwykłą przyczyną jest goroutine wysyłająca wynik do channel, którego nikt już nie czyta, bo odbiorca już poddał się po timeout. Daj goroutine wyjście, zwykle context.Context, na który też robi select, żeby mogła zrezygnować z wysłania.

Deadlock. fatal error: all goroutines are asleep - deadlock! znaczy, że każda goroutine, włącznie z main, jest zablokowana na channel lub locku, i nic już nie może jej odblokować. Najmniejszym przykładem jest wysłanie bez bufora w main bez żadnej innej działającej goroutine, która mogłaby je odebrać.

Race, których channels nie zapobiegają. Channel synchronizuje przekazanie wartości. Nie chroni zmiennej, którą dalej dotykasz poza channel. Dwie goroutines zwiększające ten sam przechwycony licznik wciąż wchodzą w race, channel albo nie, zupełnie jak w przypadku wątków opisanym w co to jest race condition i jak ją naprawić. Go ma do tego wbudowany detektor: skompiluj lub przetestuj z -race, a runtime zgłosi współbieżne dostępy do tej samej pamięci, gdzie przynajmniej jeden jest zapisem.

1
2

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

Wykrywa tylko race, które faktycznie zdarzają się podczas działania, więc znajduje ich więcej pod realistycznym współbieżnym obciążeniem niż w spokojnym teście jednostkowym. To obciążenie jest też tam, gdzie testy pełne goroutines stają się niestabilne — ten sam rodzaj niestabilności opisany w testy jednostkowe vs integracyjne vs end-to-end. Przy włączonym -race oczekuj kilka razy większego zużycia pamięci i CPU. Uruchamiaj go w CI, nie w binarce, którą wydajesz.

Channel czy mutex: co wybrać?

Obie rzeczy zabezpieczają współbieżny dostęp, ale z przeciwnych stron. sync.Mutex chroni współdzielony stan na miejscu: zablokuj, przeczytaj lub zapisz, odblokuj, a dane nigdy się nie przemieszczają. Channel przenosi własność wartości z jednej goroutine do drugiej, więc z definicji dotyka jej tylko jedna goroutine naraz. Własna wskazówka Go: używaj channels do koordynowania lub przekazywania pracy między goroutines, a mutex do ochrony jednego kawałka współdzielonych danych, który aktualizuje na miejscu kilka goroutines, jak cache czy licznik. Jeśli funkcja potrzebuje obu, prawdopodobnie robi dwie rzeczy, które lepiej by się czytały rozdzielone.

Jeśli przychodzisz z JavaScriptu, model jest inny. Jeden wątek wykonuje twój kod, a await oddaje kontrolę tylko w miejscach widocznych w kodzie źródłowym; async/await i event loop w JavaScript wyjaśnione pokazuje, gdzie leżą te punkty wstrzymania. Go daje ci prawdziwą równoległość, więc wybór prymitywu należy do ciebie.

Do kolejnego batch job (zmiana rozmiaru folderu zdjęć albo wywołanie API raz na wiersz CSV) zacznij od worker pool powyżej. Wybierz liczbę workers, zamknij channel jobs, gdy skończysz wysyłać, i dodaj context.Context do select, gdy pierwszy raz potrzebujesz anulowania.

Powiązane artykuły