golang channel

1.底层实现
2.有无缓存区别
3.线程安全
4.使用注意
5.CSP并发模型
6.使用场景

1.底层实现
channel底层是通过循环数组实现的先入先出队列，使用互斥锁保证channel写入和读取数据的线程安全

2.有无缓存区别
(1)创建方式不同, 有缓冲的chan会带上缓冲区大小，make(chan type, size)
(2)阻塞：
无缓冲的读写都会阻塞，有其他协程准备读了, channel才能写入
有缓冲的chan当缓冲满了 写入阻塞

3.线程安全
底层使用互斥锁，保证读写channel队列的线程安全

4.使用注意
(1)已经close的chan
写：panic
读：如果缓冲区还有数据可以读；如果缓冲没有数据，返回空值
close: panic (一个channel不能多次关闭)
(2)未初始化的chan
读写 死锁
关闭 panic
(3)信号通知
如果多个goroutinue都监听同一个channel,那么channel的数据被随机一个gorountine读取消费
如果多个goroutinue都监听同一个channel,这个channel被关闭,所有gorountine都能收到退出信号

5.CSP并发模型
golang基于goroutinue和channel实现csp并发模型，将生产者和消费者解耦
不要通过共享内存来通信，而是使用通信来共享内存

6.使用场景
不同协程之前通信
停止信号监听
定时任务
生产者消费者解耦
控制并发数

runtime/chan.go

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

	// chan是否关闭
	closed   uint32
	elemtype *_type // element type

	// 下一次写下标位置
	sendx uint // send index
	// 下一次读下标位置
	recvx uint // receive index

	// 读写等待goroutinue队列
	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
}


// 获取循环数组指定位置
// chanbuf(c, i) is pointer to the i'th slot in the buffer.
func chanbuf(c *hchan, i uint) unsafe.Pointer {
	return add(c.buf, uintptr(i)*uintptr(c.elemsize))
}

// 写入
func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
	// ....
	if c.qcount < c.dataqsiz {
		// Space is available in the channel buffer. Enqueue the element to send.
		qp := chanbuf(c, c.sendx)
		if raceenabled {
			racenotify(c, c.sendx, nil)
		}
		typedmemmove(c.elemtype, qp, ep)
		c.sendx++
		if c.sendx == c.dataqsiz {
			c.sendx = 0
		}
		c.qcount++
		unlock(&c.lock)
		return true
	}
	// ...
}


// 读取
func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
	// 。。。
	if c.qcount > 0 {
		// Receive directly from queue
		qp := chanbuf(c, c.recvx)
		if raceenabled {
			racenotify(c, c.recvx, nil)
		}
		if ep != nil {
			typedmemmove(c.elemtype, ep, qp)
		}
		typedmemclr(c.elemtype, qp)
		c.recvx++
		if c.recvx == c.dataqsiz {
			c.recvx = 0
		}
		c.qcount--
		unlock(&c.lock)
		return true, true
	}
	// ...
}
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
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88