在看本篇文章前我们需要了解阻塞的概念
在执行过程中暂停，以等待某个条件的触发 ，我们就称之为阻塞
在Go中我们make一个channel有两种方式，分别是有缓冲的和没缓冲的
缓冲channel 即 buffer channel 创建方式为 make(chan TYPE,SIZE)
如 make(chan int,3) 就是创建一个int类型，缓冲大小为3的 channel
非缓冲channel 即 unbuffer channel 创建方式为 make(chan TYPE)
如 make(chan int) 就是创建一个int类型的非缓冲channel
非缓冲channel 和 缓冲channel 的区别
非缓冲 channel，channel 发送和接收动作是同时发生的
例如 ch := make(chan int) ，如果没 goroutine 读取接收者<-ch ，那么发送者ch<- 就会一直阻塞
缓冲 channel 类似一个队列，只有队列满了才可能发送阻塞
代码演示
非缓冲 channel
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
package main
import (
"fmt"
"time"
)
func loop(ch chan int) {
for {
select {
case i := <-ch:
fmt.Println("this value of unbuffer channel", i)
}
}
}
func main() {
ch := make(chan int)
ch <- 1
go loop(ch)
time.Sleep(1 * time.Millisecond)
}
|
这里会报错 fatal error: all goroutines are asleep - deadlock! 就是因为 ch<-1 发送了，但是同时没有接收者，所以就发生了阻塞
但如果我们把 ch <- 1 放到 go loop(ch) 下面，程序就会正常运行
缓冲 channel
的阻塞只会发生在 channel 的缓冲使用完的情况下
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
package main
import (
"fmt"
"time"
)
func loop(ch chan int) {
for {
select {
case i := <-ch:
fmt.Println("this value of unbuffer channel", i)
}
}
}
func main() {
ch := make(chan int,3)
ch <- 1
ch <- 2
ch <- 3
ch <- 4
go loop(ch)
time.Sleep(1 * time.Millisecond)
}
|
这里也会报 fatal error: all goroutines are asleep - deadlock! ，这是因为 channel 的大小为 3 ，而我们要往里面塞 4 个数据，所以就会阻塞住
解决的办法有两个
把 channel 开大一点，这是最简单的方法，也是最暴力的
把 channel 的信息发送者 ch <- 1 这些代码移动到 go loop(ch) 下面 ，让 channel 实时消费就不会导致阻塞了
补充：3种优雅的Go channel用法
写Go的人应该都听过Rob Pike的这句话
Do not communicate by sharing memory; instead, share memory by communicating.
相信很多朋友和我一样，在实际应用中总感觉不到好处，为了用channel而用。但以我的切身体会来说，这是写代码时碰到的场景不复杂、对channel不熟悉导致的，所以希望这篇文章能给大家带来点新思路，对Golang优雅的channel有更深的认识 ：）
Fan In/Out
数据的输出有时候需要做扇出／入（Fan In／Out），但是在函数中调用常常得修改接口，而且上下游对于数据的依赖程度非常高，所以一般使用通过channel进行Fan In／Out，这样就可以轻易实现类似于shell里的管道。
|
1
2
3
4
5
6
7
8
9
10
11
12
|
func fanIn(input1, input2 <-chan string) <-chan string {
c := make(chan string)
go func() {
for {
select {
case s := <-input1: c <- s
case s := <-input2: c <- s
}
}
}()
return c
}
|
同步Goroutine
两个goroutine之间同步状态，例如A goroutine需要让B goroutine退出，一般做法如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
|
func main() {
g = make(chan int)
quit = make(chan bool)
go B()
for i := 0; i < 3; i++ {
g <- i
}
quit <- true // 没办法等待B的退出只能Sleep
fmt.Println("Main quit")
}
func B() {
for {
select {
case i := <-g:
fmt.Println(i + 1)
case <-quit:
fmt.Println("B quit")
return
}
}
}
/*
Output:
1
2
3
Main quit
*/
|
可是了main函数没办法等待B合适地退出，所以B quit 没办法打印，程序直接退出了。
然而，chan是Go里的第一对象，所以可以把chan传入chan中，所以上面的代码可以把quit 定义为chan chan bool，以此控制两个goroutine的同步
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
|
func main() {
g = make(chan int)
quit = make(chan chan bool)
go B()
for i := 0; i < 5; i++ {
g <- i
}
wait := make(chan bool)
quit <- wait
<-wait //这样就可以等待B的退出了
fmt.Println("Main Quit")
}
func B() {
for {
select {
case i := <-g:
fmt.Println(i + 1)
case c := <-quit:
c <- true
fmt.Println("B Quit")
return
}
}
}
/* Output
1
2
3
B Quit
Main Quit
*/
|
分布式递归调用
在现实生活中，如果你要找美国总统聊天，你会怎么做？
第一步打电话给在美国的朋友，然后他们也会发动自己的关系网，再找可能认识美国总统的人，以此类推，直到找到为止。
这在Kadmelia分布式系统中也是一样的，如果需要获取目标ID信息，那么就不停地查询，被查询节点就算没有相关信息，也会返回它觉得最近节点，直到找到ID或者等待超时。
好了，这个要用Go来实现怎么做呢？
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
|
func recursiveCall(ctx context.Context, id []byte, initialNodes []*node){
seen := map[string]*node{} //已见过的节点记录
request := make(chan *node, 3) //设置请求节点channel
// 输入初始节点
go func() {
for _, n := range initialNodes {
request <- n
}
}()
OUT:
for {
//循环直到找到数据
if data != nil {
return
}
// 在新的请求，超时和上层取消请求中select
select {
case n := <-request:
go func() {
// 发送新的请求
response := s.sendQuery(ctx, n, MethodFindValue, id)
select {
case <-ctx.Done():
case msg :=<-response:
seen[responseToNode(response)] = n //更新已见过的节点信息
// 加载新的节点
for _, rn := range LoadNodeInfoFromByte(msg[PayLoadStart:]) {
mu.Lock()
_, ok := seen[rn.HexID()]
mu.Unlock()
// 见过了，跳过这个节点
if ok {
continue
}
AddNode(rn)
// 将新的节点送入channel
request <- rn
}
}
}
}()
case <-time.After(500 * time.Millisecond):
break OUT // break至外层，否则仅仅是跳至loop外
case <-ctx.Done():
break OUT
}
}
return
}
|
这时的buffered channel类似于一个局部queue，对需要的节点进行处理，但这段代码的精妙之处在于，这里的block操作是select的，随时可以取消，而不是要等待或者对queue的长度有认识。
以上为个人经验，希望能给大家一个参考，也希望大家多多支持服务器之家。如有错误或未考虑完全的地方，望不吝赐教。
原文链接：https://blog.csdn.net/qq_36431213/article/details/83281250