go高并发数据结构---channel

1. chan 是什么，为什么用它

1.1 chan 的声明

代码演示：

func main() {
	// channel 声明
	c := make(chan int)
	c <- 1 // 向管道发送 1
	<-c // 丢弃
	s := <-c // 赋给s
	
	// 或者声明string
	//c := make(chan string)
	//c <- "1"
}
1
2
3
4
5
6
7
8
9
10
11

说明：

1. make(chan int) // 表示无缓冲chan
2. make(chan bool, 0) // 无缓冲
3. make(chan string， 2) // 有缓冲， 有数字就代表有缓冲

1.2 chan 的基本用法

1. ch <- x // 发送数据
2. x = <- ch // 接收数据，赋给x
3. <- ch // 接收数据，并丢弃

1.3 错误用法及修改

1. 无缓冲区，会卡住

func main() {
	c := make(chan string)
	c <- "ping"
	fmt.Println(<-c)
}
1
2
3
4
5

2. 修改，添加协程

func main() {
	c := make(chan string)
	
	// 还必须在前面开启这个协程
	go func() {
		fmt.Println(<-c)
	}()
	c <- "ping"
}
1
2
3
4
5
6
7
8
9

1.4 内存与通信

在go官方有这样一个建议：

不要通过共享内存的方式进行通信，而是应该通过通信的方式共享内存

怎么理解呢， 我分别给出两个例子就可以了解了

1. 使用共享内存方案

func watch(p *int) {
	for true {
		if *p == 1 {
		fmt.Println("break....")
		break
		}
	}
} 
func main() {
	i := 0
	go watch(&i)
	time.Sleep(time.Second * 2)
	i = 1
	time.Sleep(time.Second)
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

共享内存方案： i的地址给另一个协程， 典型的共享内存需要一直遍历， 消耗系统资源

2. 使用通信方式

func watch(c chan int) {
	// 管道不需要一遍一遍去查询
	if <-c == 1 {
		fmt.Println("hello....")
	}
}
func main() {
	c := make(chan int)
	go watch(c)
	time.Sleep(time.Second * 2)
	c <- 1
	time.Sleep(time.Second)
}
1
2
3
4
5
6
7
8
9
10
11
12
13

通信方式： 不需要一直遍历， 提高系统使用资源

1.5 为什么使用通信来共享内存

1. 避免协程竞争和数据冲突的问题（加锁又会影响性能）
2. 更高级的抽象，降低开发难度，增加程序可读性
3. 模块之间更容易解耦，增强扩展性和可维护性

2. chan 底层设计

2.1 chan 数据结构需要有哪些成员

总共需要三个成员

1. 需要一个缓存buffer
   

2. 发送数据满之后，发送会阻塞，这时会有一个等待队列

3. 接收数据也是同理，需要有一个接收队列
   

合并起来，chan 设计需要3个重要的成员

2.2 chan 源码阅读

2.2.1 在runtime/chan.go 中

hchan 数据结构就是go官方实现的chan 底层设计

type hchan struct {
	qcount uint // total data in the queue
	dataqsiz uint // size of the circular queue
	buf unsafe.Pointer // points to an array of dataqsiz elements
	elemsize uint16
	closed uint32
	elemtype *_type // element type
	sendx uint // send index
	recvx uint // receive index
	recvq waitq // list of recv waiters
	sendq waitq // list of send waiters
	// lock protects all fields in hchan, as well as several
	// fields in sudogs blocked on this channel.
	//
	// Do not change another G's status while holding this lock
	// (in particular, do not ready a G), as this can deadlock
	// with stack shrinking.
	lock mutex
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

其中下面几个成员组成了一个缓存区环形缓存

qcount uint // total data in thequeue
dataqsiz uint // size of the circularqueue
buf unsafe.Pointer // points to an arrayof dataqsiz elements
elemsize uint16
closed uint32
elemtype *_type // element type
1
2
3
4
5
6

下面两个成员组成了发送队列，其中是用的链表实现的

从这里看出是形成的链表

同理这两个成员组成了接收队列

另外还有一个互斥锁， 这个字段是锁所有字段的

下面是对hchan 中锁的认识 :

1. 互斥锁并不是排队发送、接收数据
2. 互斥锁保护的hchan 结构体本身
3. channel并不是无锁的

最后还有一个状态值字段， close 字段
1 ： 表开启
0 ：表关闭

2.3 chan 成员一览表

channel 数据结构是go 中的一等公民

chan 成员一览表：

3. chan 收发数据原理

3.1 channel 发送数据的底层原理

上面讲解了数据结构，这里就是算法了

3.1.1 <- 关键字

1. c <- 关键字是一个语法糖
2. 编译阶段，会把c <- 转化为 runtime.chansend1() // 可以用汇编看
3. chansend1() 会调用chansend() 方法

3.1.2 channel 发送的情形

1. 直接发送
2. 放入缓存
3. 休眠等待

3.1.3 直接发送

1. 发送数据前，已经有G 在休眠等待 （之前写的代码就是这种 watch）
2. 此时缓存肯定是空的，不用考虑缓存
3. 将数据直接拷贝给G的接收量， 唤醒G

code:

func watch(c chan int) {
	// 管道不需要一遍一遍去查询
	if <-c == 1 {
	fmt.Println("hello....")
	}
} 
func main() {
	c := make(chan int)
	go watch(c)
	time.Sleep(time.Second * 2)
	c <- 1
	time.Sleep(time.Second)
}
1
2
3
4
5
6
7
8
9
10
11
12
13

实现步骤：

1. 从队列里取出一个等待接收的G
2. 将数据直接拷贝到接收变量中
3. 唤醒G

3.1.4 放入缓存

1. 没有G 在休眠等待，但是有缓存空间
2. 将数据放入缓存

实现步骤：

1. 获取可存入的缓存地址
2. 存入数据
3. 维护索引 （告诉下一位存在什么位置）

3.1.5 休眠等待

原理：

1. 没有G在休眠等待，而且没有缓存或满了
2. 自己进入发送队列，休眠等待

实现步骤：
3. 把自己包装成sudog
4. sudog 放入sendq 队列
5. 休眠并解锁
6. 被唤醒后，数据已经被取走，维护其他数据

3.1.6 channel 发送数据总结

编译阶段， 会把 <- 转化为 runtime.chansend1()

1. 直接发送时，将数据直接拷贝到目标变量
2. 放入缓存时，将数据放入环形缓存，成功返回
3. 休眠等待时，将自己包装后放入sendq， 休眠

3.2 channel 接收数据原理

3.2.1 <- 关键字

1. <- c 关键字是一个语法糖
2. 编译阶段， i <- c 转化为 runtime.chanrecv1()
3. 编译阶段， i， ok <- c 转化为runtime.chanrecv2()
4. 上面两种最后会调用chanrecv() 方法

3.2.2 channel 接收的情形

接收步骤：

1. 有等待的G， 从G接收
2. 有等待的G， 从缓存接收
3. 接收缓存
4. 阻塞缓存

3.2.3 有等待的G， 从G 接收

1. 原理
   

• 接收数据前，已经有G 在休眠等待发送
• 而且这个channel 没有缓存
• 将数据直接从G 拷贝过来，唤醒G

2. 实现

• 判断有G 在发送队列等待，进入recv()
• 判断此channel 无缓存
• 直接从等待的G中取走数据，唤醒G

3.2.4 有等待的G， 从缓存接收

原理

1. 接收数据前，已经有G 在休眠等待发送
2. 而且这个channel 有缓存
3. 从缓存取走一个数据

实现

1. 判断有G 在发送队列等待，进入recv（）
2. 判断此channel 有缓存
3. 从缓存中取走一个数据
4. 将G的数据放入缓存，唤醒G

3.2.5 接收缓存

原理

1. 没有G在休眠等待发送，但是缓存有内容
2. 从缓存取走数据

实现

1. 判断没有G 在发送队列等待
2. 判断此channel有缓存
3. 从缓存中取走一个数据

3.2.6 阻塞接收

原理

1. 没有G 在休眠等待，而且没有缓存或缓存空
2. 自己进入接收队列，休眠等待

实现
3. 判断没有G在发送队列等待
4. 判断此channel 无缓存
5. 将自己包装成sudog
6. sudog 放入接收等待队列，休眠
7. 唤醒时，发送的G已经把数据拷贝到位

3.2.7 channel 接收数据总结

1. 编译阶段， <- c 会转化为chanrecv（）
2. 有等待的G， 且无缓存时，从G接收
3. 有等待的G， 且有缓存时，从缓存接收
4. 无等待的G，且缓存有数据，从缓存接收
5. 无等待的G，且缓存无数据，等待喂数据

4. 非阻塞的channel 怎么做

4.1 select 提出

只用 <- 接收时， 永远是阻塞的，什么时候用非阻塞呢， 答案是 select

4.2 select 原理

1. 同时存在接收、发送、默认路径
2. 首先查看是否有可以即使执行的case
3. 没有的话，有default， 走default
4. 没有default， 把自己注册在所有channel 中，休眠的等待

code：

func main() {
	// 快捷键补全 ctrl + alt + v （qq音乐会冲突）
	c1 := make(chan int, 5)
	c2 := make(chan int)
	select {
		case <-c1:
		fmt.Println("c1")
		case c2 <- 1:
		fmt.Println("c2")
		default:
		fmt.Println("none")
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13

4.3 timer 包的使用

1. timer 可以提供一个channel， 定时塞入数据

// 倒计时 定时器
t := time.NewTimer(time.Second)
// 倒计数1s 后定时器的channel 会塞入一个数据， 这时就可以取出一个数据了
<-t.C
fmt.Println("hello...")
1
2
3
4
5

5. 总结

5.1 为什么使用channel

1. 相对于无锁 （避免协程竞争和数据冲突的问题）
2. 相对于加锁

• 更高级的抽象，降低开发难度，增加程序的可读性
• 模块之间更容易解耦，增强扩展性和可维护性

5.2 channel 基本结构

1. 一个环形缓存
2. 两个链表（发送协程、接收协程）
3. 一个互斥锁（保护hchan）
4. 一个状态值
   

5.3 channel数据发送原理

1. 直接发送时，将数据直接拷贝到目标变量
2. 放入缓存时，将数据放入环形缓存，成功返回
3. 休眠等待时，将自己包装后放入sendq， 休眠

5.4 channel数据接收原理

1. 有等待的G，且无缓存时，从G接收
2. 有等待的G，且有缓存时，从缓存接收
3. 无等待的G，且缓存有数据，从缓存接收
4. 无等待的G， 且缓存无数据，等待喂数据

5.5 非阻塞channel

1. 使用select 可以使用channel 的非阻塞特性
2. 使用timer 配合select 可以实现超时特性