Channel trong Golang: buffered, unbuffered, close và range

Gửi và nhận

Channel là một đường ống có kiểu nối giữa các goroutine. ch <- v để gửi, <-ch để nhận. Tạo channel bằng 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)
}

Giá trị zero của kiểu channel là nil, nên var ch chan string mà không có make sẽ cho bạn một channel chặn mãi mãi. Luôn tạo channel bằng make.

Channel unbuffered dùng để đồng bộ

make(chan T) tạo một channel unbuffered. Một lần gửi sẽ chặn cho đến khi có bên nhận lấy giá trị, và một lần nhận sẽ chặn cho đến khi có bên gửi đưa giá trị tới. Hai goroutine gặp nhau tại điểm đó, nên channel unbuffered vừa là công cụ đồng bộ vừa là đường ống dữ liệu. Khi <-done ở dưới trả về, bạn biết worker đã làm xong mọi việc trước lần gửi của nó:

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

Ở đây đọc result trong main là an toàn mà không cần mutex. Mô hình bộ nhớ của Go bảo đảm mọi thứ worker làm trước lần gửi đều nhìn thấy được trong main sau lần nhận tương ứng. chan struct{} là kiểu chuẩn cho một tín hiệu thuần túy, vì struct{} không chiếm bộ nhớ.

Channel buffered

make(chan T, n) cho channel chỗ chứa n giá trị. Gửi vẫn thành công khi chưa có bên nhận, cho đến khi buffer đầy; nhận thành công cho đến khi buffer rỗng. Giá trị đi ra theo đúng thứ tự đi vào.

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

Buffer tách rời bên gửi và bên nhận để những đợt dữ liệu ngắn không làm bên gửi bị đứng. Nó không chữa được một producer luôn nhanh hơn consumer; nó chỉ lùi thời điểm bên gửi bị chặn. Hãy chọn kích thước buffer có lý do (số bên gửi, một kích thước lô đã biết), đừng chọn chỉ để deadlock biến mất.

len(ch) chỉ là ảnh chụp tại một thời điểm. Đến lúc bạn dùng nó, goroutine khác có thể đã làm nó thay đổi, nên đừng dùng nó để đoán một lần gửi có bị chặn hay không. Hãy dùng select với nhánh default cho việc đó.

Close và range

close(ch) báo cho bên nhận rằng sẽ không còn giá trị nào được gửi nữa. Sau khi đóng:

• các giá trị đã nằm trong buffer vẫn được giao,
• sau đó mọi lần nhận trả về giá trị zero ngay lập tức,
• v, ok := <-ch cho kết quả ok == false,
• for v := range ch kết thúc.

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

Lần nhận đầu tiên sau close vẫn lấy được "last" với ok == true. Lần thứ hai nhận giá trị zero "" và false.

Các quy tắc gây panic:

• gửi vào channel đã đóng gây panic với send on closed channel,
• đóng một channel đã đóng gây panic,
• đóng một channel nil gây panic.

Vì vậy chỉ phía gửi mới đóng channel, và chỉ đóng một lần. Khi có nhiều bên gửi, không bên gửi nào biết khi nào những bên khác xong việc; hãy để một goroutine riêng chờ tất cả bên gửi (bằng sync.WaitGroup) rồi đóng channel sau khi Wait trả về. Chỉ cần đóng channel khi có bên nhận chờ điểm kết thúc. Một channel chưa đóng mà không ai tham chiếu tới sẽ được thu gom rác như mọi giá trị khác.

Kiểu có chiều

Một hàm có thể khai báo rằng nó chỉ gửi hoặc chỉ nhận trên một channel. Khi đó trình biên dịch từ chối thao tác còn lại.

Một chan T tự động chuyển thành một trong hai kiểu bị giới hạn khi bạn truyền nó vào hàm. Đó là lý do produce ở trên có thể trả về <-chan int: bên gọi duyệt range được nhưng không gửi vào hay đóng được. Hãy dùng kiểu có chiều cho mọi tham số hàm phù hợp. Chúng ghi rõ ai sở hữu channel và biến việc dùng sai thành lỗi biên dịch.

Deadlock

Nếu mọi goroutine đều bị chặn và không gì đánh thức được chúng, runtime sẽ dừng chương trình:

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

Bản dump goroutine cho bạn biết thao tác nào bị kẹt (ở đây là chan send, hoặc chan receive, sync.WaitGroup.Wait, select). Các nguyên nhân thường gặp:

• gửi vào channel unbuffered khi không có bên nhận nào đang chạy,
• range qua một channel không bao giờ được đóng,
• bộ đếm WaitGroup không bao giờ về 0,
• hai goroutine chờ lẫn nhau.

Runtime chỉ phát hiện trường hợp tất cả goroutine đều bị kẹt. Trong một máy chủ còn các goroutine khác đang sống (một HTTP listener, một ticker), cùng lỗi đó không làm gì sập cả; goroutine bị chặn chỉ đơn giản bị rò rỉ.

Channel nil

Gửi và nhận trên một channel nil sẽ chặn mãi mãi. Nghe có vẻ vô dụng, nhưng đây là một mẹo chuẩn bên trong select: gán biến channel thành nil sẽ vô hiệu hóa nhánh của nó. Đoạn code này gộp hai channel và ngừng lắng nghe từng channel khi nó đã đóng:

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

Nếu không có các phép gán nil, một channel đã đóng luôn ở trạng thái sẵn sàng, và vòng lặp sẽ quay mãi trên các giá trị zero.

Một pipeline

Các channel ghép lại thành pipeline: mỗi giai đoạn là một goroutine nhận từ một channel và gửi sang channel tiếp theo, rồi đóng đầu ra của mình khi đầu vào đã hết.

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

Mỗi giai đoạn chạy đồng thời, và thứ tự đầu ra là xác định (1, 16, 81) vì mỗi giai đoạn là một goroutine duy nhất giữ nguyên thứ tự. Các lần đóng lan truyền theo dây chuyền: generate đóng, làm vòng range của square kết thúc, square đóng đầu ra của nó, và cứ thế xuống đến main.

Điểm yếu của pipeline này: nếu main ngừng đọc sớm, các giai đoạn sẽ bị chặn ở lần gửi mãi mãi. Pipeline thực tế nhận thêm một context.Context hoặc một channel done và dùng select trên nó cạnh mỗi lần gửi.

Channel hay mutex

Channel dùng để chuyển quyền sở hữu dữ liệu và để báo hiệu sự kiện. Một sync.Mutex đơn giản hơn khi cần bảo vệ trạng thái dùng chung mà nhiều goroutine đọc và cập nhật tại chỗ, như một cache hay một bộ đếm. Một struct có mutex bên trong thường rõ ràng hơn một goroutine giữ trạng thái và phục vụ các yêu cầu qua channel. Hãy chọn cách nào khiến code ngắn hơn và quyền sở hữu rõ ràng hơn.

Tra cứu nhanh