Go —— channel （二）

news2024/11/23 1:20:55

一个空的 channel 会产生哪些问题

读写nil管道均会阻塞触发死锁。关闭的管道仍然可以读取数据，向关闭的管道写数据会触发panic。

问：如果有多个协程同时读取一个channel，channel会如何选择消费者

channel 会按照维护的 recvq 等待读消息的协程队列按照FIFO的顺序选择消费者

我们先来看一下 channel 源码

type hchan struct {
	qcount   uint           // 当前队列中剩余元素个数
	dataqsiz uint           // 环形队列长度，即可以存放的元素个数
	buf      unsafe.Pointer // 环形队列指针 缓冲区
	elemsize uint16			// 每个元素的大小
	closed   uint32			// 标识关闭状态
	elemtype *_type // 元素类型
	sendx    uint   // 队列下标，指示元素写入时存放到队列中的位置
	recvx    uint   // 队列下标，指示下一个被读取元素在队列中的位置
	recvq    waitq  // 等待读消息的协程队列
	sendq    waitq  // 等待写消息的协程队列
	lock mutex		// 互斥锁，chan不允许并发读写
}

向管道写数据

向一个管道中写数据的简单过程如下

如果缓冲区中有空余位置，则将数据写入缓冲区，结束发送过程
如果缓冲区中没有空余位置，则将当前协程加入sendq队列，进入休眠并等待被读协程唤醒

简单流程如下图所示

在这里插入图片描述

向管道读数据

channel 会维护一个等待读消息的协程队列 recvq，当一个协程读取消息时的简单过程如下：

如果缓冲区中有数据，则从缓冲区中取出数据，结束读取过程
如果缓冲区中没有数据，则将当前协程加入 recvq 队列，进入休眠并等待被写协程唤醒

如果 sendq 不为空，且没有缓冲区，则会从 sendq队列的第一个协程中获取数据

简单流程如下图所示：

在这里插入图片描述

编写一个程序，测试一下

func main() {
	c := make(chan int)
	wg := sync.WaitGroup{}
	wg.Add(100)
	go func() {				// G1
		for {
			a := <-c
			fmt.Println("1", a)
			wg.Done()
		}
	}()
	go func() {				// G2
		for {
			a := <-c
			fmt.Println("2", a)
			wg.Done()
		}
	}()
	go func() {				// G3
		for {
			a := <-c
			fmt.Println("3", a)
			wg.Done()
		}
	}()
	go func() {				// G4
		for {
			a := <-c
			fmt.Println("4", a)
			wg.Done()
		}
	}()
    time.Sleep(1 * time.Second)		// 等待四个协程排好队
	for i := 0; i < 100; i++ {
		c <- i
	}
	wg.Wait()
}

按照上面协程读取消息的过程会发生什么呢？

当c还未被写入消息时， G1~G4 以FIFO的原则排好队，比如现在的recvq的顺序为 G1、G2、G3、G4
当主协程发送消息时，无缓冲区，直接从recvq 队列中取出头部G ，随后再添加到队尾
c中的数据按照G1、G2、G3、G4的顺序依次被读取

那实际执行结果是怎样的呢

在我多次测试后发现前四个数会被每个协程消费一次，随后会出现大片数据被同一协程消费的情况

3 2					// 前四次
3 4
3 5
3 6
3 7
3 8
3 9
3 10
3 11
3 12
3 13
3 14
3 15
3 16
3 17
3 18
3 19
3 20
3 21
3 22
3 23
3 24
3 25
3 26
3 27
3 28
3 29
3 30
3 31
3 32
3 33
3 34
3 35
3 36
3 37
3 38
3 39
3 40
3 41
3 42
3 43
3 44
3 45
3 46
3 47
3 48
3 49
3 50
3 51
3 52
3 53
3 54
3 55
3 56
3 57
3 58
3 59
3 60
3 61
3 62
3 63
3 64
3 65
3 66
3 67
3 68
3 69
3 70
3 71
3 72
3 73
3 74
3 75
3 76
3 77
3 78
3 79
3 80
3 81
3 82
3 83
3 84
3 85
3 86
3 87
3 88
3 89
3 90
3 91
3 92
3 93
3 94
3 95
3 96
3 97
3 98
3 99
2 1					// 前四次
1 0					// 前四次
4 3					// 前四次

Process finished with the exit code 0

出现这种情况可能与GMP调度模型有关系，当我们继续增大数据量后（比如增加到10000），会发现每个协程读取chan的次数其实差不多。

当我们向管道写数据时添加一个间隔时间

func main() {
	c := make(chan int)
	wg := sync.WaitGroup{}
	wg.Add(100)
	go func() {				// G1
		for {
			a := <-c
			fmt.Println("1", a)
			wg.Done()
		}
	}()
	go func() {				// G2
		for {
			a := <-c
			fmt.Println("2", a)
			wg.Done()
		}
	}()
	go func() {				// G3
		for {
			a := <-c
			fmt.Println("3", a)
			wg.Done()
		}
	}()
	go func() {				// G4
		for {
			a := <-c
			fmt.Println("4", a)
			wg.Done()
		}
	}()
    time.Sleep(1 * time.Second)		// 等待四个协程排好队
	for i := 0; i < 100; i++ {
       	time.Sleep(1 * time.Millisecond)	// 每隔1ms 发送一次
		c <- i
	}
	wg.Wait()
}