Goroutine'ler, kanallar ve sync

Bir kafede tek bir barista varsa sıra yavaş ilerler: kahveler tek tek hazırlanır. Üç barista aynı anda çalıştığında siparişler çok daha çabuk hazır olur. Go programları da böyle çalışabilir: tek bir program binlerce görevi aynı anda yürütür. Go'yu bulut sunucuları için popüler yapan da bu basit eşzamanlılık modelidir: goroutine'ler, kanallar ve sync paketi.

Goroutine'ler

Bir goroutine, Go çalışma zamanının kendisi tarafından yönetilen hafif bir görevdir. Başlatmak için bir fonksiyon çağrısının önüne go yazmak yeterlidir: go work(). Bir goroutine yalnızca birkaç kilobaytlık bellekle başlar; bu yüzden binlerce, hatta milyonlarca goroutine oluşturulabilir. Önemli bir kural: main biter bitmez program da biter ve diğer goroutine'leri beklemez.

Goroutine'lerin bitmesini beklemek için sync.WaitGroup kullanılır: Add(1) sayacı artırır, her goroutine sonunda Done() çağırır, Wait() ise sayaç sıfıra inene kadar bekler. Aşağıdaki örnekte her goroutine sonucunu dilimin kendi indeksine yazar; bu yüzden birbirlerine engel olmazlar ve yazdırma sırası hep aynıdır.

package main

import (
	"fmt"
	"sync"
)

func square(n int) int {
	return n * n
}

func main() {
	nums := []int{2, 3, 4, 5}
	results := make([]int, len(nums))
	var wg sync.WaitGroup
	for i, n := range nums {
		wg.Add(1)
		go func() {
			defer wg.Done()
			results[i] = square(n)
		}()
	}
	wg.Wait()
	fmt.Println(results)
}

[4 9 16 25]

Kanallar

Bir kanal, goroutine'ler arasında değer taşıyan bir borudur: ch := make(chan int). ch <- v bir değer gönderir, v := <-ch ise bir değer alır. Arabelleksiz bir kanalda gönderen, alıcı hazır olana kadar bekler; bu, goroutine'leri otomatik olarak eşitler. Gönderen işini bitirince close(ch) çağırır, alıcı ise kanal kapanana kadar for v := range ch ile okur. chan<- int yalnızca gönderme, <-chan int ise yalnızca alma kanalıdır.

package main

import "fmt"

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

func main() {
	ch := make(chan int)
	go producer(ch)
	for v := range ch {
		fmt.Println("received", v)
	}

	done := make(chan string)
	go func() {
		done <- "work finished"
	}()
	fmt.Println(<-done)
}

received 10
received 20
received 30
work finished

Arabellekli bir kanalın (make(chan string, 3)) içinde yer vardır: arabellek dolana kadar gönderen beklemez. Bir kuyruk gibi çalışır: ilk giren ilk çıkar. len kanalda bekleyen eleman sayısını, cap ise arabellek boyutunu verir. Kapalı ve boş bir kanaldan okumak hemen sıfır değeri ve ok = false döndürür.

package main

import "fmt"

func main() {
	queue := make(chan string, 3)
	queue <- "Aysel"
	queue <- "Murad"
	fmt.Println(len(queue), cap(queue))
	fmt.Println(<-queue)
	queue <- "Leyla"
	close(queue)
	for name := range queue {
		fmt.Println("next:", name)
	}
	v, ok := <-queue
	fmt.Printf("%q %v\n", v, ok)
}

2 3
Aysel
next: Murad
next: Leyla
"" false

select ve zaman aşımları

select birkaç kanal işlemini aynı anda bekler ve hangisi önce hazır olursa onu çalıştırır; switch'e benzer ama kanallar içindir. time.After(d) belirli bir süre sonra değer veren bir kanal döndürür; bu yüzden onunla zaman aşımı kurmak kolaydır. Bir default kolu eklersen select hiç beklemez: hazır kanal yoksa hemen default çalışır.

package main

import (
	"fmt"
	"time"
)

func main() {
	fast := make(chan string)
	go func() {
		time.Sleep(10 * time.Millisecond)
		fast <- "result ready"
	}()
	select {
	case msg := <-fast:
		fmt.Println(msg)
	case <-time.After(time.Second):
		fmt.Println("timeout")
	}

	slow := make(chan string)
	select {
	case msg := <-slow:
		fmt.Println(msg)
	case <-time.After(50 * time.Millisecond):
		fmt.Println("timeout")
	}
}

result ready
timeout

Mutex'ler ve işçi havuzu

Birkaç goroutine aynı değişkeni değiştirdiğinde bir veri yarışı (data race) oluşur: c.value++ aslında üç adımdır (oku, artır, yaz) ve goroutine'ler birbirinin sonucunun üzerine yazabilir. sync.Mutex bunu önler: Lock() ile Unlock() arasındaki kodu aynı anda yalnızca bir goroutine çalıştırabilir.

package main

import (
	"fmt"
	"sync"
)

type Counter struct {
	mu    sync.Mutex
	value int
}

func (c *Counter) Inc() {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.value++
}

func main() {
	var c Counter
	var wg sync.WaitGroup
	for range 1000 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			c.Inc()
		}()
	}
	wg.Wait()
	fmt.Println(c.value)
}

1000

Gerçek projelerde binlerce görevin her biri için ayrı bir goroutine başlatmak yerine genellikle bir işçi havuzu (worker pool) kurulur: sabit sayıda işçi goroutine ortak bir jobs kanalından görev alır ve sonuçları bir results kanalına yazar. Böylece aynı anda çalışan goroutine sayısı kontrol altında kalır. Sonuçlar herhangi bir sırayla gelebilir; bu yüzden onları toplayıp sıralıyoruz.

package main

import (
	"fmt"
	"slices"
	"sync"
)

func worker(jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
	defer wg.Done()
	for n := range jobs {
		results <- n * n
	}
}

func main() {
	jobs := make(chan int, 10)
	results := make(chan int, 10)
	var wg sync.WaitGroup
	for range 3 {
		wg.Add(1)
		go worker(jobs, results, &wg)
	}
	for n := 1; n <= 6; n++ {
		jobs <- n
	}
	close(jobs)
	wg.Wait()
	close(results)

	var squares []int
	for r := range results {
		squares = append(squares, r)
	}
	slices.Sort(squares)
	fmt.Println(squares)
}

[1 4 9 16 25 36]

Sıraya dikkat et: önce tüm görevler gönderilir ve işçilerin range döngüleri bitebilsin diye jobs kapatılır. Ardından wg.Wait() bütün işçileri bekler ve ancak bundan sonra results kapatılır. results kanalının arabelleği 10 olduğu için işçiler sonuç yazarken bloke olmaz.

Belleği paylaşarak iletişim kurma; iletişim kurarak belleği paylaş.