面试准备-通道

type hchan struct {
	// chan 里元素数量
	qcount   uint
	// chan 底层循环数组的长度
	dataqsiz uint
	// 指向底层循环数组的指针
	// 只针对有缓冲的 channel
	buf      unsafe.Pointer
	// chan 中元素大小
	elemsize uint16
	// chan 是否被关闭的标志
	closed   uint32
	// chan 中元素类型
	elemtype *_type // element type
	// 已发送元素在循环数组中的索引
	sendx    uint   // send index
	// 已接收元素在循环数组中的索引
	recvx    uint   // receive index
	// 等待接收的 goroutine 队列
	recvq    waitq  // list of recv waiters
	// 等待发送的 goroutine 队列
	sendq    waitq  // list of send waiters

	// 保护 hchan 中所有字段
	lock mutex
}

type waitq struct {
	first *sudog
	last  *sudog
}

// 无缓冲通道
ch1 := make(chan int)
// 有缓冲通道
ch2 := make(chan int, 10)

func makechan(t *chantype, size int64) *hchan

const hchanSize = unsafe.Sizeof(hchan{}) + uintptr(-int(unsafe.Sizeof(hchan{}))&(maxAlign-1))

func makechan(t *chantype, size int64) *hchan {
	elem := t.elem

	// 省略了检查 channel size，align 的代码
	// ……

	var c *hchan
	// 如果元素类型不含指针 或者 size 大小为 0（无缓冲类型）
	// 只进行一次内存分配
	if elem.kind&kindNoPointers != 0 || size == 0 {
		// 如果 hchan 结构体中不含指针，GC 就不会扫描 chan 中的元素
		// 只分配 "hchan 结构体大小 + 元素大小*个数" 的内存
		c = (*hchan)(mallocgc(hchanSize+uintptr(size)*elem.size, nil, true))
		// 如果是缓冲型 channel 且元素大小不等于 0（大小等于 0的元素类型：struct{}）
		if size > 0 && elem.size != 0 {
			c.buf = add(unsafe.Pointer(c), hchanSize)
		} else {
			// race detector uses this location for synchronization
			// Also prevents us from pointing beyond the allocation (see issue 9401).
			// 1. 非缓冲型的，buf 没用，直接指向 chan 起始地址处
			// 2. 缓冲型的，能进入到这里，说明元素无指针且元素类型为 struct{}，也无影响
			// 因为只会用到接收和发送游标，不会真正拷贝东西到 c.buf 处（这会覆盖 chan的内容）
			c.buf = unsafe.Pointer(c)
		}
	} else {
		// 进行两次内存分配操作
		c = new(hchan)
		c.buf = newarray(elem, int(size))
	}
	c.elemsize = uint16(elem.size)
	c.elemtype = elem
	// 循环数组长度
	c.dataqsiz = uint(size)

	// 返回 hchan 指针
	return c
}

## 5、向channel发送数据

要了解向`c *hchan`中发送数据时需要的一些字段

​	c.qcount	通道缓存中的元素个数，无缓冲通道一直为0

​	c.datasiz	通道底层数组中可存储的元素个数

​	c.closed	通道状态标志，=0表示为关闭，=1表示关闭

​	c.sendx	下一个要发送进来的元素在循环数组中的下标位置

​	c.recvx	下一个要被接受者接收的元素在循环数组中的下标位置

​	c.recvq	等待接收消息的协程队列

​	c.sendq	等到发送消息的协程队列

一些状态

c.qcount<c.datasiz: 只在缓冲型通道中出现，说明缓冲空间没可用。

c.recvq.first == nil: 等待队列中有协程，缓存空间为空

c.sendq.first == nil: 发送队列中有协程，缓存空间满

**不阻塞的发送**：向chan中发送元素时，无需向chan中的buf中拷贝元素值，而是将消息直接拷贝到接收协程的工作栈中，这有两种情况

- a、cahnnel是非缓冲型的，这里会检测recvq中是否有等待接收的协程，如果有，直接将数据拷贝到等待协程中，没有的话就挂起发送协程
- b、channel是缓冲型的：这里如果检测到缓冲空间已经存满元素就直接挂起发送协程，如果检测到recvq中有等待的接收协程，说明通道缓存中没有元素，这时候就可以直接将元素拷贝到等待携程中去。

这种不向缓冲区中写元素，而是直接将元素发送到指定协程的栈空间中的做法的好处是可以减少一次加锁和内存拷贝的次数，性能更高。

**阻塞发送：**

- 如果 `c.qcount < c.dataqsiz`，说明缓冲区可用（肯定是缓冲型的 channel）。先通过函数取出待发送元素应该去到的位置：

  ```golang
qp := chanbuf(c, c.sendx)

  // 返回循环队列里第 i 个元素的地址处
func chanbuf(c *hchan, i uint) unsafe.Pointer {
	return add(c.buf, uintptr(i)*uintptr(c.elemsize))
}

// entry points for <- c from compiled code
func chanrecv1(c *hchan, elem unsafe.Pointer) {
	chanrecv(c, elem, true)
}

func chanrecv2(c *hchan, elem unsafe.Pointer) (received bool) {
	_, received = chanrecv(c, elem, true)
	return
}

  当我们观察到 channel 没准备好接收：

  1. 非缓冲型，等待发送列队里没有 goroutine 在等待
2. 缓冲型，但 buf 里没有元素

  之后，又观察到 closed == 0，即 channel 未关闭。此时并没由被选中，而且也没有取到数据，最后的返回值是 (false, false)。

## 7、关闭cahnnel

close 逻辑比较简单，对于一个 channel，recvq 和 sendq 中分别保存了阻塞的发送者和接收者。**关闭 channel 后，对于等待接收者而言，会收到一个相应类型的零值**。对于等待发送者，会直接 panic。所以，在不了解 channel 还有没有接收者的情况下，不能贸然关闭 channel。

close 函数先上一把大锁，接着把所有挂在这个 channel 上的 sender 和 receiver 全都连成一个 sudog 链表，再解锁。最后，再将所有的 sudog 全都唤醒。

唤醒之后，该干嘛干嘛。sender 会继续执行 chansend 函数里 goparkunlock 函数之后的代码，很不幸，检测到 channel 已经关闭了，panic。receiver 则比较幸运，进行一些扫尾工作后，返回。这里，selected 返回 true，而返回值 received 则要根据 channel 是否关闭，返回不同的值。如果 channel 关闭，received 为 false，否则为 true。这我们分析的这种情况下，received 返回 false。

## 8、从一个关闭的channel仍然能读取出数据吗？

从一个有缓冲的 channel 里读数据，当 channel 被关闭，依然能读出有效值。只有当返回的 ok 为 false 时，读出的数据才是无效的。

```go
func main() {
	ch := make(chan int, 5)
	ch <- 18
	close(ch)
	x, ok := <-ch
	if ok {
		fmt.Println("received: ", x)
	}

	x, ok = <-ch
	if !ok {
		fmt.Println("channel closed, data invalid.")
	}
}

func IsClosed(ch <-chan T) bool {
	select {
	case <-ch:
		return true
	default:
	}

	return false
}

func main() {
	c := make(chan T)
	fmt.Println(IsClosed(c)) // false
	close(c)
	fmt.Println(IsClosed(c)) // true
}

func SafeCloseChannel(ch chan int) (justClosed bool) {
defer func() {
if recover() != nil {
 justClosed = false
}
}()

 close(ch)
return true
}

type MyChannel struct{
   C chan struct{}
   once sync.Once
}

func NewMyChannel() *MyChannel{
   return &MyChannel{C:make(chan struct{})}
}

func (mc *MyChannel) SafeClose(){
   mc.once.Do(func(){
      close(mc.C)
   })
}

func main() {
	rand.Seed(time.Now().UnixNano())
	const Max = 100000
	const NumSenders = 1000
	dataCh := make(chan int, 100)
	stopCh := make(chan struct{})

	for i := 0; i < NumSenders; i++ {
		go func() {
			for {
				select {
				case <-stopCh:
					return
				case dataCh <- rand.Intn(Max):

				}
			}
		}()
	}
	go func() {
		for value := range dataCh {
			if value == Max-1 {
				fmt.Println("send stop signal to senders")
                // 当我们关闭一个cahnnel的时候，所有的receiver会受到对应类型的零值。
                // 这时候select分支case <-stopCh被选中，退出函数，不再发送数据。
				close(stopCh)
				return
			}
			fmt.Println(value)
		}
	}()
	select {
	case <-time.After(time.Hour):

	}
}

func main() {
	rand.Seed(time.Now().UnixNano())
	const Max = 100000
	const NumSenders = 1000
	const NumReceiver = 10
	dataCh := make(chan int, 100)
	stopCh := make(chan struct{})
	toStop := make(chan string, 1)

	// moderator
	go func() {
		<-toStop
		close(stopCh)
	}()

	for i := 0; i < NumSenders; i++ {
		go func(id string) {
			for {
				value := rand.Intn(Max)
				if value == 0 {
					select {
					case toStop <- "sender#" + id:
					default:
					}
					return
				}

				select {
				case <-stopCh:
					return
				case dataCh <- value:
				}
			}
		}(strconv.Itoa(i))
	}

	for i := 0; i < NumReceiver; i++ {
		go func(id string) {
			for {
				select {
				case <-stopCh:
					return
				case value := <-dataCh:
					if value == Max-1 {
						select {
						case toStop <- "receiver#" + id:
						default:
						}
						return
					}
					fmt.Println(value)
				}
			}
		}(strconv.Itoa(i))
	}
	select {
	case <-time.After(time.Hour):
	}
}

type user struct {
	name string
	age  int
}

var u = user{name: "Ankur", age: 25}
var g = &u

func modifyUser(pu *user) {
	fmt.Println("modifyUser Received Vaule", pu)
	pu.name = "Anand"
}

func printUser(u <-chan *user) {
	time.Sleep(2 * time.Second)
	fmt.Println("printUser goRoutine called", <-u)
}

func main() {
	c := make(chan *user, 5)
	c <- g
	fmt.Println(g)
	// modify g
	g = &user{name: "Ankur Anand", age: 100}
	go printUser(c)
	go modifyUser(g)
	time.Sleep(5 * time.Second)
	fmt.Println(g)
}

&{Ankur 25}
modifyUser Received Vaule &{Ankur Anand 100}
printUser goRoutine called &{Ankur 25}
&{Anand 100}

select{
    case <- time.After(100 * time.Millisecond):
    case <- s.stopc:
    	return false
}

func worker() {
	ticker := time.Tick(1 * time.Second)
	for {
		select {
		case <- ticker:
			// 执行定时任务
			fmt.Println("执行 1s 定时任务")
		}
	}
    ticker := time.Tick(1 * time.Second)
	// 创建一个信号通道
	sigs := make(chan os.Signal, 1)
	// 监听 SIGINT 和 SIGTERM 信号
	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)
	var i = 0
	for {
		select {
		case <-ticker:
            // 执行定时任务
			fmt.Println(i)
		case <-sigs:
            // 结束定时任务
			fmt.Println("shutdown!!!")
			return
		}
		i++
	}
}

func worker(taskCh <-chan int) {
	const N = 5
	for i := 0; i < N; i++ {
		go func(id int) {
			for {
				task := <-taskCh
				fmt.Println("finish task: %d by worker %d\n", task, id)
				time.After(1 * time.Second)
			}
		}(i)
	}
}

func main() {
	// 创建一个信号通道
	sigs := make(chan os.Signal, 1)
	// 监听 SIGINT 和 SIGTERM 信号
	signal.Notify(sigs, syscall.SIGINT, syscall.SIGTERM)

	taskCh := make(chan int, 100)
	go worker(taskCh)

	for i := 0; i < 10; i++ {
		taskCh <- i
	}

	select {
	case <-time.After(time.Hour):
	case <-sigs:
		fmt.Println("shutdown!!!")
		return
	}
}

finish task: %d by worker %d
 0 4
finish task: %d by worker %d
 5 4
finish task: %d by worker %d
 6 4
finish task: %d by worker %d
 7 4
finish task: %d by worker %d
 8 4
finish task: %d by worker %d
 9 4
finish task: %d by worker %d
 2 0
finish task: %d by worker %d
 3 2
finish task: %d by worker %d
 4 3
finish task: %d by worker %d
 1 1

var limit = make(chan int, 3)

func main() {
    // …………
    for _, w := range work {
        go func() {
            limit <- 1
            w()
            <-limit
        }()
    }
    // …………
}