Golang Channel là gì? Các ứng dụng Channel trong Golang

Golang là ngôn ngữ hỗ trợ cho việc lập trình concurrency một cách đơn giản và hiệu quả với Goroutines. Để giao tiếp bộ nhớ giữa các goroutines, chúng ta có một triết lý rất nổi tiếng:

Do not communicate by sharing memory; instead, share memory by communicating - Rob Pike

"Đừng giao tiếp bằng việc chia sẻ bộ nhớ, thay vào đó hãy chia sẻ bộ nhớ thông qua việc giao tiếp". "Việc giao tiếp" có nghĩa là trong Golang chúng ta sẽ có một kênh giao tiếp dữ liệu được thiết kế riêng. Đó chính là Channel.

Vì để đơn giản và tập trung, bài viết này sẽ chỉ đề cập đến unbuffered channel trong Golang.

Golang Channel là gì?

Channel là các kênh giao tiếp trung gian giữa các Goroutines trong Golang. Channel giúp các goroutines có thể gởi và nhận được dữ liệu cho nhau một cách an toàn thông qua cơ chế lock-free.

Mặc định, Channel là kênh giao tiếp 2 chiều. Nghĩa là Channel có thể dùng cho cả gởi và nhận dữ liệu.

Cách khai báo Channel thông qua từ khoá chan trong Golang

Để sử dụng Channel, chúng ta dùng keyword chan được hỗ trợ mặc định từ trong chính ngôn ngữ Golang (thường được hiểu là first-class). Chúng ta không cần phải import thêm bất kì một package nào để sử dụng chan.

Syntax khai báo Channel với chan:

var channelName chan Type

Hoặc

channelName := make(chan Type)

Vì Golang là ngôn ngữ Strong Typed, tức là phải có kiểu dữ liệu cụ thể cho các biến, việc này cũng không ngoại lệ với Channel. Channel phải biết được kiểu dữ liệu gì sẽ đi qua nó.

Gởi và nhận dữ liệu qua Channel

Để gởi và nhận dữ liệu qua Channel, chúng ta sẽ dùng toán tử <-. Toán tử này hoạt động như một chỉ hướng dữ liệu sẽ đi từ đâu đến đâu. Chỉ hướng này giúp ta xác định được dữ liệu đang được gởi đi hay nhận về.

Gởi dữ liệu vào Channel

channelName <- value

Việc gởi dữ liệu vào Channel sẽ giống như: "Tôi đã hoàn tất công việc của mình với dữ liệu này và bàn giao chúng cho người khác".

Nhận dữ liệu từ Channel

myVar := <- channelName

Ví dụ gởi và nhận dữ liệu với Channel:

package main

import "fmt"

func main() {
	myChan := make(chan int)

	go func() {
		myChan <- 1
	}()

	fmt.Println(<-myChan)
}

Cơ chế block của Channel

Việc gởi và nhận dữ liệu thông qua Channel sẽ có hỗ trợ cơ chế block. Việc này giúp các Goroutines giao tiếp qua Channel một cách đồng bộ (synchronize). Về nguyên tắc, Channel sẽ block goroutines nếu nó chưa sẵn sàng.

Chúng ta cần hiểu rằng giao tiếp dữ liệu qua Channel giống như việc ta phải "trao tận tay" cho người nhận. Nếu vì lý do nào đó, người nhận chưa sẵn sàng, hoặc ngược lại người trao chưa đến, đôi bên sẽ phải "đợi".

Đoạn code dưới đây sẽ khiến main goroutine bị lock lại vĩnh viễn, gây lỗi deadlock và chương trình sẽ exit.

package main

func main() {
	myChan := make(chan int)
	myChan <- 1 // deadlock here
}

Main goroutine đang gởi giá trị 1 vào channel myChan. Bản thân main goroutine sẽ bị block lại cho tới khi có goroutine khác nhận dữ liệu từ  myChan. Chương trình bị deadlock vì ngoài main ra chẳng còn goroutine nào lấy dữ liệu ra từ myChan cả.

Các goroutines dù có chạy nhanh chậm như thế nào, khi giao tiếp với Channel đều phải bị "dừng" lại để giao tiếp rồi muốn làm gì thì làm. 200Lab sẽ minh hoạ thêm một ví dụ nữa:

package main

import (
	"fmt"
	"time"
)

func main() {
	myChan := make(chan int)

	go func() {
		for i := 1; i <= 10; i++ {
			myChan <- i
			time.Sleep(time.Second)
		}
	}()

	for i := 1; i <= 10; i++ {
		fmt.Println(<-myChan)
	}
}

Trong đoạn code trên chúng ta sẽ có 2 goroutine là main và một function vô danh. Main goroutine sẽ lấy dữ liệu từ myChan với tốc độ cao hơn function gởi vào. Ở mỗi vòng lặp, main goroutine cũng sẽ phải đợi cho tới khi myChan có dữ liệu gởi vào. Kết quả ta sẽ thấy các con số được in ra sau mỗi giây.

Thật ra đây chính là cơ chế "streaming" rất hữu dụng về sau này. Golang đã làm nó trở nên cực kì gọn và dễ dùng.

Cách sử dụng Channel chỉ gởi hoặc nhận

Sử dụng Channel cho cả gởi và nhận dữ liệu

Mặc định, Channel sẽ dùng cho cả 2 chiều giao tiếp gởi và nhận như sau:

package main

import (
	"fmt"
)

func receiveAndSend(c chan int) {
	fmt.Printf("Received: %d\n", <-c)
	fmt.Printf("Sending 2...\n")
	c <- 2
}

func main() {
	myChan := make(chan int)

	go receiveAndSend(myChan)
	myChan <- 1

	fmt.Printf("Value from receiveAndSend: %d\n", <-myChan)
}

Trong đoạn code trên, hàm receiveAndSend thực hiện lấy dữ liệu ra và ngay sau đó là gởi dữ liệu vào c. Như vậy channel c dùng cho cả gởi và nhận dữ liệu (hay giao tiếp 2 chiều).

Tuy nhiên chúng ta vẫn có thể giới hạn lại channel chỉ được dùng cho gởi hoặc nhận ở một hàm nào đó.

Sử dụng Channel chỉ được phép nhận dữ liệu

Để giới hạn Channel ở một hàm nào đó chỉ được phép nhận dữ liệu, không thể gởi vào thì chúng ta sẽ dùng <-chan:

func recieveOnly(c <-chan int) {
	fmt.Printf("Received: %d\n", <-c)
	c <- 2 // error
}

Sẽ có lỗi biên dịch nếu ta cố tình gởi dữ liệu vào Channel trong hàm recieveOnly

Sử dụng Channel chỉ được phép gởi dữ liệu

Tương tự, nếu cần giới hạn Channel ở một hàm nào đó chỉ được dùng để gởi dữ liệu ta sẽ dùng chan<-:

func sendOnly(c chan<- int) {
	c <- 2 // OK
	fmt.Printf("Received: %d\n", <-c) // error
}

Lỗi biên dịch sẽ xảy ra nếu ta cố tình lấy dữ liệu từ channel trong hàm sendOnly.

Close Channel trong Golang

Định nghĩa và cú pháp hàm close

Để đóng một Channel chúng ta sẽ dùng hàm close(). Khi một Channel bị close thì có nghĩa là không còn dữ liệu nào sẽ đi qua nó nữa.

Close() cũng là một first-class function của Golang. Cú pháp như sau:

close(chanName)

Các lưu ý khi close channel trong Golang

1. Không thể gởi dữ liệu và channel đã bị close, sẽ có lỗi runtime nếu ta cố làm việc này.
2. Khi hàm close() thực thi bản chất sẽ truyền vào channel một tín hiệu để phía sử dụng biết là channel đã close hay chưa.

Kiểm tra một Channel đã bị đóng hay chưa

Bình thường nếu chỉ lấy dữ liệu từ channel đơn thuần thì chúng ta không biết được channel đã đóng hay vẫn còn hoạt động. Vì thế một syntax bổ sung để làm được việc này như sau:

value, isAlive := <-chanName

Ví dụ minh hoạ:

package main

import (
	"fmt"
)

func main() {
	myChan := make(chan int)

	go func() {
		for i := 1; i <= 10; i++ {
			myChan <- i
		}
		close(myChan)
	}()

	for {
		value, isAlive := <-myChan

		if !isAlive {
			fmt.Printf("Value: %d. Channel has been closed.\n", value)
			break
		}

		fmt.Printf("Value: %d\n", value)
	}
}

Value: 1
Value: 2
Value: 3
Value: 4
Value: 5
Value: 6
Value: 7
Value: 8
Value: 9
Value: 10
Value: 0. Channel has been closed.

Như đã lưu ý ở trên, nếu chúng ta sử dụng chan int và không dùng biến check isAlive thì trường hợp value = 0 sẽ bị hiểu sai là có giá trị 0 được gởi qua channel. Hãy nhớ rằng các biến number trong Golang sẽ luôn là zero (0) nếu không được gán (empty).

Sử dụng for-range với channel

Một cách tiện lợi hơn rất nhiều để vừa lấy dữ liệu từ channel vừa có thể biết được channel còn hoạt động hay không với cú pháp for-range như sau:

package main

import (
	"fmt"
)

func main() {
	myChan := make(chan int)

	go func() {
		for i := 1; i <= 10; i++ {
			myChan <- i
		}
		close(myChan)
	}()

	for value := range myChan {
		fmt.Printf("Value: %d\n", value)
	}
}

Sử dụng for-range như trên sẽ ngắn gọn và tiện lợi hơn rất đáng kể, hạn chế lỗi logic check channel. Rất nhiều trường hợp lỗi if-else fix muốn xỉu luôn!!

Sử dụng Select và WaitGroup để xử lý nhiều channel

Khi có nhiều hơn 1 channel cùng được sử dụng sẽ bắt đầu phát sinh những vấn đề:

1. Làm sao để biết channel nào về dữ liệu trước hoặc sẵn sàng nhận dữ liệu để ưu tiên xử lý?!
2. Làm sao để biết tất cả channel đều đã về dữ liệu hoặc đóng?!

200Lab sẽ dùng một ví dụ đơn giản như sau:

package main

import (
	"fmt"
	"math/rand"
	"time"
)

func main() {
	r := rand.New(rand.NewSource(time.Now().Unix()))

	ch1 := make(chan int)
	ch2 := make(chan int)

	go func() {
		time.Sleep(time.Second * time.Duration(r.Intn(5)))
		ch1 <- 1
	}()

	go func() {
		time.Sleep(time.Second * time.Duration(r.Intn(5)))
		ch2 <- 2
	}()

	fmt.Println(<-ch1)
    fmt.Println(<-ch2)
}

Đoạn code trên có 2 channel đặt trong 2 goroutines thực thi random thời gian nên ta không biết channel nào sẽ có dữ liệu sớm hơn channel còn lại. Cách viết trên sẽ luôn đợi để in ra ch1 rồi đến ch1 dù thực tế có khả năng ch2 nhanh hơn.

Sử dụng Select để chọn Channel đã sẵn sàng

Theo định nghĩa, Select sẽ block cho tới khi có case  có thể được thực thi trên các channel. Vì thế ta có thể ứng dụng như sau:

package main

import (
	"fmt"
	"math/rand"
	"time"
)

func main() {
	r := rand.New(rand.NewSource(time.Now().Unix()))

	ch1 := make(chan int)
	ch2 := make(chan int)

	go func() {
		time.Sleep(time.Second * time.Duration(r.Intn(5)))
		ch1 <- 1
	}()

	go func() {
		time.Sleep(time.Second * time.Duration(r.Intn(5)))
		ch2 <- 2
	}()

	select {
	case v1 := <-ch1:
		fmt.Println("Ch1 come first with value:", v1)
		fmt.Println("then ch2 with value:", <-ch2)
	case v2 := <-ch2:
		fmt.Println("Ch2 come first with value:", v2)
		fmt.Println("then ch1 with value:", <-ch1)
	}
}

Select trong thực tế hầu như sẽ đi với for tạo cặp syntax for-select như trong ví dụ từ Go.dev:

package main

import "fmt"

func fibonacci(c, quit chan int) {
	x, y := 0, 1
	for {
		select {
		case c <- x:
			x, y = y, x+y
		case <-quit:
			fmt.Println("quit")
			return
		}
	}
}

func main() {
	c := make(chan int)
	quit := make(chan int)
	go func() {
		for i := 0; i < 10; i++ {
			fmt.Println(<-c)
		}
		quit <- 0
	}()
	fibonacci(c, quit)
}

For-select xuất hiện trong hàm fibonacci để tạo một loop xét trên 2 case select là:

1. Case 1: Channel c có thể tiếp nhận dữ liệu x (không bị block), thực hiện phép toán x = y và y = x + y.
2. Case 2: Channel quit có dữ liệu, return thoát hàm luôn.

Tại main goroutine, có một loop 10 lần lấy dữ liệu channel c. Việc này khiến case 1 sẽ được thực thi 10 lần. Sau đó channel quit được truyền dữ liệu vào để case 2 thực thi (lúc này case 1 bị block, không thể thực thi được).

Phương thức sử dụng 1 channel riêng để quit, thoát hàm hoặc vòng lặp này trong practice xuất hiện rất phổ biến, có thể được xem là một best practice.

Sử dụng Waitgroup để biết các goroutine đã hoàn tất

Đây sẽ là một trường hợp rất phổ biến vì chúng ta sẽ không biết khi nào thì các goroutine hay channel đã hoàn tất. Trong thực tế ta sẽ không bao giờ biết được là sleep bao lâu. Ví dụ cho vấn đề này như sau:

package main

import (
	"fmt"
	"math/rand"
	"time"
)

func main() {
	r := rand.New(rand.NewSource(time.Now().Unix()))

	go func() {
		time.Sleep(time.Second * time.Duration(r.Intn(5)))
		fmt.Println("Goroutine 1 done.")
	}()

	go func() {
		time.Sleep(time.Second * time.Duration(r.Intn(5)))
		fmt.Println("Goroutine 2 done.")
	}()

	time.Sleep(time.Second * 6)
}

Để sử dụng waitgroup chúng ta cần import package "sync". Nguyên tắc hoạt động của WaitGroup khá đơn giản là có bao nhiêu goroutines cần đợi thì dùng hàm Add(number). Mỗi khi goroutines chạy xong thì gọi hàm Done(). Hàm Wait() sẽ bị block cho tới khi đã done hết tất cả số lượng goroutines đã khai báo trước đó.

package main

import (
	"fmt"
	"math/rand"
	"sync"
	"time"
)

func main() {
	r := rand.New(rand.NewSource(time.Now().Unix()))
	wc := new(sync.WaitGroup)
	wc.Add(2)

	go func() {
		time.Sleep(time.Second * time.Duration(r.Intn(5)))
		fmt.Println("Goroutine 1 done.")
		wc.Done()
	}()

	go func() {
		time.Sleep(time.Second * time.Duration(r.Intn(5)))
		fmt.Println("Goroutine 2 done.")
		wc.Done()
	}()

	wc.Wait()
	fmt.Println("All Goroutines done")
}

Một số ví dụ sử dụng Channel trong Golang

Sau đây là một số ví dụ minh hoạ Channel, không phải các best practices.

Sử dụng 1 channel để lắng nghe dữ liệu từ nhiều nơi

package main

import (
	"fmt"
	"runtime"
)

func sender(c chan<- int, name string) {
	for i := 1; i <= 100; i++ {
		c <- 1
		fmt.Printf("%s has sent 1 to channel\n", name)
		runtime.Gosched()
	}
}

func main() {
	myChan := make(chan int)

	go sender(myChan, "S1")
	go sender(myChan, "S2")
	go sender(myChan, "S3")

	start := 0

	for {
		start += <-myChan
		fmt.Println(start)

		if start >= 300 {
			break
		}
	}
}

Cân bằng tải với Channel

package main

import (
	"fmt"
	"time"
)

func publisher() <-chan int {
	c := make(chan int)

	go func() {
		for i := 1; i <= 1000; i++ {
			c <- i
		}

		close(c)
	}()

	return c
}

func consumer(c <-chan int, name string) {
	counter := 0

	for value := range c {
		fmt.Printf("Consumer %s is doing task %d\n", name, value)
		counter++
		time.Sleep(time.Millisecond * 20)
	}

	fmt.Printf("Consumer %s has finished %d task(s)\n", name, counter)
}

func main() {
	myChan := publisher()
	maxConsumer := 5

	for i := 1; i <= maxConsumer; i++ {
		go consumer(myChan, fmt.Sprintf("%d", i))
	}

	time.Sleep(time.Second * 10)
}

Ghép kênh (merge channel) sử dụng waitgroup và channel

package main

import (
	"fmt"
	"sync"
)

func streamNumbers(numbers ...int) <-chan int {
	c := make(chan int)

	go func() {
		for n := range numbers {
			c <- n
		}

		close(c)
	}()

	return c
}

func sumAllStreams(streams ...<-chan int) <-chan int {
	sumChan := make(chan int)
	counter := 0
	wc := new(sync.WaitGroup)

	wc.Add(len(streams))

	for i := 0; i < len(streams); i++ {
		go func(s <-chan int) {
			for n := range s {
				counter += n
			}
			wc.Done()
		}(streams[i])
	}

	go func() {
		wc.Wait()
		sumChan <- counter
	}()

	return sumChan
}

func main() {
	s := sumAllStreams(
		streamNumbers(1, 2, 3, 4, 5),
		streamNumbers(8, 8, 3, 3, 10, 12, 14),
		streamNumbers(1, 1, 2, 2, 4, 4, 6),
	)

	fmt.Println(<-s)
}

Lời kết

Đây là tất cả những gì bạn cần biết về Channel, chính xác là unbuffered channel. 200Lab đã có một bài viết riêng về buffered channel cũng như so sánh với unbuffered channel.

Ngoài ra thì trong khoá học Golang của 200Lab cũng có chia sẻ rất nhiều best practice về Channel Golang. Chúc các bạn học tốt!

Bài viết liên quan:

• Golang là gì? Backend Developer có nên học Golang
• Goroutines là gì? Lập trình concurrency chưa bao giờ dễ như Golang
• Những sai lầm thường thấy khi bạn sử dụng goroutines
• Buffered Channel là gì? Thường xuyên bị hỏi trong phỏng vấn Golang Dev