Go中怎么使用channel

這篇文章將為大家詳細講解有關Go中怎么使用channel，小編覺得挺實用的，因此分享給大家做個參考，希望大家閱讀完這篇文章后可以有所收獲。

一、使用channel等待任務結束

使用案例還是在第一篇的第二節中寫的代碼，不過這里只需要一段即可。

package mainimport (
	"fmt"
	"time")func createWorker(id int) chan<- int {
	c := make(chan int)
	go worker(id, c)
	return c}func worker(id int, c chan int) {
	for n := range c {
		fmt.Printf("Worker %d receive %c\n", id, n)
	}}func channelDemo() {
	var channels [10]chan<- int
	for i := 0; i < 10; i++ {
		channels[i] = createWorker(i)
	}

	for i := 0; i < 10; i++ {
		channels[i] <- 'a' + i	}

	for i := 0; i < 10; i++ {
		channels[i] <- 'A' + i	}
	time.Sleep(time.Millisecond)}func main() {
	channelDemo()}

這里咔咔將原始源碼放在這里，如果你想跟著文章的節奏走，可以放到你的編輯器中進行操作。

那這段代碼的問題是在哪里呢？

可以看到在channelDemo函數最后使用了一個sleep，這玩意在程序中可不能亂用。

說到這里給大家講一個小故事，咔咔之前在網上看到一段就是加了sleep的代碼。

然后一個新手程序員不明白為什么要加這個sleep，然后問題項目經理，項目經理說老板發現程序慢之后會找咱們優化，每一次優化把這個sleep的時間縮短即可。讓老板感覺到我們在做事情。

新手就是新手對不懂得代碼都會進行標注，然后就寫了一句注釋“項目經理要求這里運行緩慢，老板讓優化時，代碼得到明顯的速度提升”。

這句話很不巧的是被老板給看見了，老板不認識代碼，但文字還是認識的哈！于是，項目經理下馬。

所以說對于sleep大多數都是一個測試狀態，堅決不會出現在線上的，所以呢？就要解決代碼中的這個sleep。

那么大家在回憶一下，在這里為什么要加sleep呢？

發送到channel的數據都是在另一個goroutine中進行并發打印的，并發打印就會出現問題，因為根本不會知道什么時候才打印完畢。

所以說這個sleep就會為了應對這個不知道什么時候打印完的問題，給個1毫秒讓進行打印。

這種做法是非常不好的，接下來看看使用一種新的方式來解決這個問題。

以下代碼是修改完的代碼。

package mainimport (
	"fmt")type worker struct {
	in   chan int
	done chan bool}func createWorker(id int) worker {
	w := worker{
		in:   make(chan int),
		done: make(chan bool),
	}
	go doWorker(id, w.in, w.done)
	return w}func doWorker(id int, c chan int, done chan bool) {
	for n := range c {
		fmt.Printf("Worker %d receive %c\n", id, n)
		done <- true
	}}func channelDemo() {
	var workers [10]worker	for i := 0; i < 10; i++ {
		workers[i] = createWorker(i)
	}

	for i := 0; i < 10; i++ {
		workers[i].in <- 'a' + i		<-workers[i].done	}

	for i := 0; i < 10; i++ {
		workers[i].in <- 'A' + i		<-workers[i].done	}}func main() {
	channelDemo()}

將這些代碼復制到你的本地，然后再來看一下都做了什么改動。

• 首先為了參數傳遞方便，建立了一個結構體worker

• 并且把之前的worker方法改為了doWorker

• 這個時候createWorker方法返回值就不能是之前的channel了，而是創建的結構體worker

• 然后在createWorker方法里邊把channel全部創建好。并且使用結構體給doWorker傳遞參數。

• 最終返回的就是結構體。

• 最后一步就是給channelDemo方法里邊發送數據的倆個循環里邊接收一下workers[i]的值即可。

看一下打印結果

是不是有點懵，這怎么成有序的了，如果是并行的那還有必要開那10個worker，直接按照順序打印就好了。

現在就來解決這個問題，我不希望發一個任務然后等它結束。

最好的就是把他們全部發出去，等待它們全部結束再退出來。

代碼實現如下

package mainimport (
	"fmt")type worker struct {
	in   chan int
	done chan bool}func createWorker(id int) worker {
	w := worker{
		in:   make(chan int),
		done: make(chan bool),
	}
	go doWorker(id, w.in, w.done)
	return w}func doWorker(id int, c chan int, done chan bool) {
	for n := range c {
		fmt.Printf("Worker %d receive %c\n", id, n)
		done <- true
	}}func channelDemo() {
	var workers [10]worker	for i := 0; i < 10; i++ {
		workers[i] = createWorker(i)
	}

	for i, worker := range workers {
		worker.in <- 'a' + i	}

	for i, worker := range workers {
		worker.in <- 'A' + i	}

	for _, worker := range workers {
		<-worker.done		<-worker.done	}}func main() {
	channelDemo()}

在這里再進行打印看一下結果，你會發現代碼是有問題的。

為什么將小寫的字母打印出來，而打印大寫字母時發生了報錯呢？

這個就要追溯到代碼中了，因為我們代碼本身就寫的有問題。

還是回歸到本文長談的一個問題，那就是對于所有的channel有發送數據就必須有接收數據，如果沒有接收數據就會報錯。

那么在代碼中你能看出是那塊只進行了發送數據，而沒有接收數據嗎？

這個問題就是當給channel把小寫字母發送了后，就會到進入到doWorker方法，然后給done發送了一個true，但是接收done的方法是在后面，也就是說第二個發送大寫字母時，就會發送循環的等待。

解決這個問題也很簡單，我們只需要并發的發送done即可。

看到打印結果也是正確的。

本文給的這個案例在一般項目中是不會出現的，所以說不用糾結于此。

給的案例就是為了讓大家更熟悉channel的機制而已。

對于這個解決方法還有一個方案解決，請看代碼。

將代碼還原到之前，然后在每一個發送字母的下面循環接收done即可。

對于這種多任務等待方式在go中有一個庫是可以來做這個事情，接下來看一下。

sync.WaitGroup的用法

對于sync.WaitGroup的用法咔咔就不一一介紹了，簡單的看一下源碼的實現即可。

package mainimport (
	"fmt"
	"sync")type worker struct {
	in chan int
	wg *sync.WaitGroup}func createWorker(id int, wg *sync.WaitGroup) worker {
	w := worker{
		in: make(chan int),
		wg: wg,
	}
	go doWorker(id, w.in, wg)
	return w}func doWorker(id int, c chan int, wg *sync.WaitGroup) {
	for n := range c {
		fmt.Printf("Worker %d receive %c\n", id, n)
		wg.Done()
	}}func channelDemo() {
	var wg sync.WaitGroup	var workers [10]worker	for i := 0; i < 10; i++ {
		workers[i] = createWorker(i, &wg)
	}
	// 添加20個任務
	wg.Add(20)
	for i, worker := range workers {
		worker.in <- 'a' + i	}
	for i, worker := range workers {
		worker.in <- 'A' + i	}
	wg.Wait()}func main() {
	channelDemo()}

這份源碼也是非常簡單的，具體修改得東西咔咔簡單介紹一下。

• 首先取消了channelDemo這個方法中關于done的channel。

• 使用了sync.WaitGroup，并且給createWorker方法傳遞sync.WaitGroup

• createWorker方法使用了 worker的結構體。

• 所以要先修改worker結構體，將之前的done改為wg *sync.WaitGroup即可

• 這樣就可以直接用結構體的數據。

• 接著在doWorker方法中把最后一個參數done改為wg *sync.WaitGroup

• 將方法中的done改為wg.Done()

• 最后一步就是回到函數channelDemo中把任務數添加進去，然后在代碼最后添加一個等待即可。

關于這塊的內容先知道這么用即可，咔咔后期會慢慢的補充并且深入。

抽象代碼

這塊的代碼看起來不是那么的完美的，接下來抽象一下。

這塊代碼有沒有發現有點蹩腳，接下來我們使用函數式編程進行簡單的處理。

package mainimport (
	"fmt"
	"sync")type worker struct {
	in   chan int
	done func()}func createWorker(id int, wg *sync.WaitGroup) worker {
	w := worker{
		in: make(chan int),
		done: func() {
			wg.Done()
		},
	}
	go doWorker(id, w)
	return w}func doWorker(id int, w worker) {
	for n := range w.in {
		fmt.Printf("Worker %d receive %c\n", id, n)
		w.done()
	}}func channelDemo() {
	var wg sync.WaitGroup	var workers [10]worker	for i := 0; i < 10; i++ {
		workers[i] = createWorker(i, &wg)
	}
	// 添加20個任務
	wg.Add(20)
	for i, worker := range workers {
		worker.in <- 'a' + i	}
	for i, worker := range workers {
		worker.in <- 'A' + i	}
	wg.Wait()}func main() {
	channelDemo()}

這塊代碼看不明白就先放著，寫的時間長了，你就會明白其中的含義了，學習東西不要鉆牛角尖。

二、使用select進行調度

開頭先給一個問題，假設現在有倆個channel，誰來的快先收誰應該怎么做？

package mainimport (
	"fmt"
	"math/rand"
	"time")func generator() chan int {
	out := make(chan int)
	go func() {
		i := 0
		for {
			// 隨機睡眠1500毫秒以內
			time.Sleep(
				time.Duration(rand.Intn(1500)) *
					time.Millisecond)
			// 往out這個channel發送i值
			out <- i
			i++
		}
	}()
	return out}func main() {
	// 這里需要明白如果代碼為var c1, c2 chan int  則c1和c2都為nil
	// 在 select里面也是可以使用的，只不過是堵塞狀態！
	var c1, c2 = generator(), generator()
	for {
		/**
		select 方式進行調度
		        使用場景：比如有多個通道，但我打算是哪一個通道先給我數據，我就先執行誰
		        這個select 可以是并行執行 channel管道
		*/
		select {
		case n := <-c1:
			fmt.Printf("receive from c1 %d\n", n)
		case n := <-c2:
			fmt.Printf("receive from c2 %d\n", n)
		}
	}}

以上就是代碼實現，代碼注釋也寫的非常的清晰明了，就不過多的做解釋了。

主要用法還是對channel的使用，在帶上了一個新的概念select，可以在多個通道，那個通道先發送數據，就先執行誰，并且這個select也是可以并行執行channel管道。

在上文寫的createWorker和worker倆個方法還記得吧！接下來就不在select里邊直接打印了。

就使用之前寫的倆個方法融合在一起，咔咔已將將源碼寫好了，接下來看一下實現。

package mainimport (
	"fmt"
	"math/rand"
	"time")func worker(id int, c chan int) {
	for n := range c {
		fmt.Printf("Worker %d receive %d\n", id, n)
	}}func createWorker(id int) chan<- int {
	c := make(chan int)
	go worker(id, c)
	return c}func generator() chan int {
	out := make(chan int)
	go func() {
		i := 0
		for {
			// 隨機睡眠1500毫秒以內
			time.Sleep(
				time.Duration(rand.Intn(1500)) *
					time.Millisecond)
			// 往out這個channel發送i值
			out <- i
			i++
		}
	}()
	return out}func main() {
	// 這里需要明白如果代碼為var c1, c2 chan int  則c1和c2都為nil
	// 在 select里面也是可以使用的，只不過是堵塞狀態！
	var c1, c2 = generator(), generator()
	// 直接調用createWorker方法，返回的就是一個channel
	w := createWorker(0)
	for {
		/**
		select 方式進行調度
		        使用場景：比如有多個通道，但我打算是哪一個通道先給我數據，我就先執行誰
		        這個select 可以是并行執行 channel管道
		*/
		select {
		case n := <-c1:
			w <- n		case n := <-c2:
			w <- n		}
	}}

運行代碼

看到運行結果得知也是沒有問題的。

這段代碼雖然運行沒有任何問題，但是這樣有什么缺點呢？

可以看下這段代碼n := <-c1:這里先收了一個值，然后在下邊代碼w <- n又會阻塞住，這個是不好的。

那么希望是怎么執行的呢？

這種模式是在select中既可以收數據，也可以發數據，目前這個程序是編譯不過的，請看修改后的源碼。

package mainimport (
	"fmt"
	"math/rand"
	"time")func worker(id int, c chan int) {
	for n := range c {
		fmt.Printf("Worker %d receive %d\n", id, n)
	}}func createWorker(id int) chan<- int {
	c := make(chan int)
	go worker(id, c)
	return c}func generator() chan int {
	out := make(chan int)
	go func() {
		i := 0
		for {
			// 隨機睡眠1500毫秒以內
			time.Sleep(
				time.Duration(rand.Intn(1500)) *
					time.Millisecond)
			// 往out這個channel發送i值
			out <- i
			i++
		}
	}()
	return out}func main() {
	// 這里需要明白如果代碼為var c1, c2 chan int  則c1和c2都為nil
	// 在 select里面也是可以使用的，只不過是堵塞狀態！
	var c1, c2 = generator(), generator()
	// 直接調用createWorker方法，返回的就是一個channel
	var worker = createWorker(0)
	// 這個n如果放在for循環里邊，就會一直打印0，因為從c1和c2收數據需要時間，所以會把0直接傳給worker
	n := 0
	// 使用這個標識告訴有沒有值
	hasValue := false
	for {
		// 利用nil  channel的特性
		var activeWorker chan<- int
		if hasValue {
			activeWorker = worker		}
		/**
		select 方式進行調度
		        使用場景：比如有多個通道，但我打算是哪一個通道先給我數據，我就先執行誰
		        這個select 可以是并行執行 channel管道
		*/
		select {
		case n = <-c1:
			// 收到值的話就標記為true
			hasValue = true
		case n = <-c2:
			// 收到值的話就標記為true
			hasValue = true
		case activeWorker <- n:
			hasValue = false
		}
	}}

這個模式還是有缺點的，因為n收c1和c2的速度跟消耗的速度是不一樣的。

假設c1的生成速度特別快，一下子生成了1,2，3。那么最后輸出的數據有可能就只有3，而1和2就無法輸出了。

這個場景也是非常好模擬的，只需要在打印的位置加上一點延遲時間即可。

此時你會看到運行結果為0、7、12、20…中間很多的數字都沒來得急打印。

因此我們就需要把收到的n存下來進行排隊輸出。

package mainimport (
	"fmt"
	"math/rand"
	"time")func worker(id int, c chan int) {
	for n := range c {
		// 手動讓消耗速度變慢
		time.Sleep(5 * time.Second)
		fmt.Printf("Worker %d receive %d\n", id, n)
	}}func createWorker(id int) chan<- int {
	c := make(chan int)
	go worker(id, c)
	return c}func generator() chan int {
	out := make(chan int)
	go func() {
		i := 0
		for {
			// 隨機睡眠1500毫秒以內
			time.Sleep(
				time.Duration(rand.Intn(1500)) *
					time.Millisecond)
			// 往out這個channel發送i值
			out <- i
			i++
		}
	}()
	return out}func main() {
	// 這里需要明白如果代碼為var c1, c2 chan int  則c1和c2都為nil
	// 在 select里面也是可以使用的，只不過是堵塞狀態！
	var c1, c2 = generator(), generator()
	// 直接調用createWorker方法，返回的就是一個channel
	var worker = createWorker(0)
	// 用來收n的值
	var values []int
	for {
		// 利用nil  channel的特性
		var activeWorker chan<- int
		var activeValue int
		// 判斷當values中有值時
		if len(values) > 0 {
			activeWorker = worker			// 取出索引為0的值
			activeValue = values[0]
		}
		/**
		select 方式進行調度
		        使用場景：比如有多個通道，但我打算是哪一個通道先給我數據，我就先執行誰
		        這個select 可以是并行執行 channel管道
		*/
		select {
		case n := <-c1:
			// 將收到的數據存到values中
			values = append(values, n)
		case n := <-c2:
			// 將收到的數據存到values中
			values = append(values, n)
		case activeWorker <- activeValue:
			// 送出去后就需要把values中的第一個值拿掉
			values = values[1:]
		}
	}}

以上就是實現代碼

此時在來看運行結果。

運行結果沒有漏掉數據，并且也是無序的，這樣就非常好了。

計時器的使用

上面的這個程序是退出不了的，我們想讓它10s后就直接退出怎么做呢？

那就需要使用計時器來進行操作了。

package mainimport (
	"fmt"
	"math/rand"
	"time")func worker(id int, c chan int) {
	for n := range c {
		// 手動讓消耗速度變慢
		time.Sleep(time.Second)
		fmt.Printf("Worker %d receive %d\n", id, n)
	}}func createWorker(id int) chan<- int {
	c := make(chan int)
	go worker(id, c)
	return c}func generator() chan int {
	out := make(chan int)
	go func() {
		i := 0
		for {
			// 隨機睡眠1500毫秒以內
			time.Sleep(
				time.Duration(rand.Intn(1500)) *
					time.Millisecond)
			// 往out這個channel發送i值
			out <- i
			i++
		}
	}()
	return out}func main() {
	// 這里需要明白如果代碼為var c1, c2 chan int  則c1和c2都為nil
	// 在 select里面也是可以使用的，只不過是堵塞狀態！
	var c1, c2 = generator(), generator()
	// 直接調用createWorker方法，返回的就是一個channel
	var worker = createWorker(0)
	// 用來收n的值
	var values []int
	// 返回的是一個channel
	tm := time.After(10 * time.Second)
	for {
		// 利用nil  channel的特性
		var activeWorker chan<- int
		var activeValue int
		// 判斷當values中有值時
		if len(values) > 0 {
			activeWorker = worker			// 取出索引為0的值
			activeValue = values[0]
		}
		/**
		select 方式進行調度
		        使用場景：比如有多個通道，但我打算是哪一個通道先給我數據，我就先執行誰
		        這個select 可以是并行執行 channel管道
		*/
		select {
		case n := <-c1:
			// 將收到的數據存到values中
			values = append(values, n)
		case n := <-c2:
			// 將收到的數據存到values中
			values = append(values, n)
		case activeWorker <- activeValue:
			// 送出去后就需要把values中的第一個值拿掉
			values = values[1:]
		case <-tm:
			fmt.Println("Bye")
			return
		}
	}}

這里就是源碼的實現，可以看到直接在select中是可以收到tm的值的，也就說如果到了10s，就會執行打印bye的操作。

那么現在還有另外一個需求，就是如果在800毫秒的時間內還沒有收到數據，可以做其它事情。

使用舉一反三的思想，你可以思考一下這件事情應該怎么做。

其實也就很簡單了，只需要在case中在設置一個定時器即可。

既然說到了這里就在給大家補充一個用法tick := time.Tick(time.Second)

同樣也是在case中使用。

這樣就可以每秒來顯示一下values隊列有多少數據。

這塊的內容就結束了，最終給大家發一下源碼，感興趣的可以在自己的編輯器上試試看。

package mainimport (
	"fmt"
	"math/rand"
	"time")func worker(id int, c chan int) {
	for n := range c {
		// 手動讓消耗速度變慢
		time.Sleep(time.Second)
		fmt.Printf("Worker %d receive %d\n", id, n)
	}}func createWorker(id int) chan<- int {
	c := make(chan int)
	go worker(id, c)
	return c}func generator() chan int {
	out := make(chan int)
	go func() {
		i := 0
		for {
			// 隨機睡眠1500毫秒以內
			time.Sleep(
				time.Duration(rand.Intn(1500)) *
					time.Millisecond)
			// 往out這個channel發送i值
			out <- i
			i++
		}
	}()
	return out}func main() {
	// 這里需要明白如果代碼為var c1, c2 chan int  則c1和c2都為nil
	// 在 select里面也是可以使用的，只不過是堵塞狀態！
	var c1, c2 = generator(), generator()
	// 直接調用createWorker方法，返回的就是一個channel
	var worker = createWorker(0)
	// 用來收n的值
	var values []int
	// 返回的是一個channel
	tm := time.After(10 * time.Second)
	tick := time.Tick(time.Second)
	for {
		// 利用nil  channel的特性
		var activeWorker chan<- int
		var activeValue int
		// 判斷當values中有值時
		if len(values) > 0 {
			activeWorker = worker			// 取出索引為0的值
			activeValue = values[0]
		}
		/**
		select 方式進行調度
		        使用場景：比如有多個通道，但我打算是哪一個通道先給我數據，我就先執行誰
		        這個select 可以是并行執行 channel管道
		*/
		select {
		case n := <-c1:
			// 將收到的數據存到values中
			values = append(values, n)
		case n := <-c2:
			// 將收到的數據存到values中
			values = append(values, n)
		case activeWorker <- activeValue:
			// 送出去后就需要把values中的第一個值拿掉
			values = values[1:]
		case <-time.After(800 * time.Millisecond):
			// 如果在800毫秒沒有收到數據則提示超時
			fmt.Println("timeout")
		case <-tick:
			// 每秒獲取一下values中隊列的長度
			fmt.Println("queue len = ", len(values))
		case <-tm:
			fmt.Println("Bye")
			return
		}
	}}

關于“Go中怎么使用channel”這篇文章就分享到這里了，希望以上內容可以對大家有一定的幫助，使各位可以學到更多知識，如果覺得文章不錯，請把它分享出去讓更多的人看到。