您好,登錄后才能下訂單哦!
這篇文章將為大家詳細講解有關Go中怎么使用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的用法咔咔就不一一介紹了,簡單的看一下源碼的實現即可。
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()}
這塊代碼看不明白就先放著,寫的時間長了,你就會明白其中的含義了,學習東西不要鉆牛角尖。
開頭先給一個問題,假設現在有倆個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”這篇文章就分享到這里了,希望以上內容可以對大家有一定的幫助,使各位可以學到更多知識,如果覺得文章不錯,請把它分享出去讓更多的人看到。
免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。