【并发编程】如果用 Channel 解决并发问题？

ChannelType = ("chan" | "chan" "<-" | "<-" "chan") ElementType

chan string           // 可以发送接收string
chan<- struct{}       // 只能发送struct{}
<-chan int            // 只能从chan接收int

chan<- chan int
chan<- <-chan int
<-chan <-chan int
chan (<-chan int)

chan<- chan int   // 《- 和第一个chan结合
chan<- <-chan int // 第一个<-和最左边的chan结合，第二个<-和左边第二个chan结合
<-chan <-chan int // 第一个<-和最左边的chan结合，第二个<-和左边的第二个chan结合
chan (<-chan int) // 因为括号的原因，<-和括号内第一个chan结合

meke(chan int, 9527)

ch <- 2000

x := <-ch // 把接收的一条数据赋值给变量x
foo(<-ch) // 把接收的一个的数据作为参数传给函数
<-ch // 丢弃接收的一条数据

func main() {
    var ch = make(chan int, 10)
    for i := 0; i < 10; i++ {
        select {
            case ch <- i:
            case v := <-ch:
                fmt.Println(v)
        }
    }
}

for v := range ch {
    fmt.Println(v)
}

for range ch {
}

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
}

func makechan(t *chantype, size int) *hchan {
    elem := t.elem
    // 略去检查代码
    mem, overflow := math.MulUintptr(elem.size, uintptr(size))
    
    var c *hchan
    switch {
    case mem == 0:
        // chan的size或者元素的size是0，不必创建buf
        c = (*hchan)(mallocgc(hchanSize, nil, true))
        c.buf = c.raceaddr()
    case elem.ptrdata == 0:
        // 元素不是指针，分配一块连续的内存给hchan数据结构和buf
        c = (*hchan)(mallocgc(hchanSize+mem, nil, true))
        // hchan数据结构后面紧接着就是buf
        c.buf = add(unsafe.Pointer(c), hchanSize)
    default:
        // 元素包含指针，那么单独分配buf
        c = new(hchan)
        c.buf = mallocgc(mem, elem, true)
    }
    // 元素大小、类型、容量都记录下来
    c.elemsize = uint16(elem.size)
    c.elemtype = elem
    c.dataqsiz = uint(size)
    lockInit(&c.lock, lockRankHchan)
    
    return c
}

func chansend1(c *hchan, elem unsafe.Pointer) {
   chansend(c, elem, true, getcallerpc())
}

func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {
   // 第一部分
   if c == nil {
      if !block {
         return false
      }
      gopark(nil, nil, waitReasonChanSendNilChan, traceEvGoStop, 2)
      throw("unreachable")
   }
   ......
}

// 第二部分，如果chan没有被close，并且chan满了，直接返回
if !block && c.closed == 0 && full(c) {
    return false
}

// 第三部分，chan已经被close的情景
lock(&c.lock) // 开始加锁
if c.closed != 0 {
    unlock(&c.lock)
    panic(plainError("send on closed channel"))
}

// 第四部分，从接收队列中出队一个等待的receiver
if sg := c.recvq.dequeue(); sg != nil {
    send(c, sg, ep, func() { unlock(&c.lock) }, 3)
    return true
}

// 第五部分，buf还么满
if c.qcount < c.dataqsiz {
    qp := chanbuf(c, c.sendx)
    if raceenabled {
        raceacquire(qp)
        racerelease(qp)
    }
    typedmemmove(c.elemtype,qp,ep)
    c.sendx++
    if c.sendx == c.dataqsiz {
        c.sendx = 0
    }
    c.qcount++
    unlock(&c.lock)
    return true
}

// 第六部分，buf满。
// chansend1不会进入if块里，因为chansend1的block=true
if !block {
    unlock(&c.lock)
    return false
}
......

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
}

func chanrecv(c *hchan, ep unsafe.Pointer, block bool) (selected, received bool) {
    // 第一部分，chan为nil
    if c == nil {
        if !block {
            return
        }
        gopark(nil, nil, waitReasonChanReceiveiveNilChan, traceEvGoStop, 2)
        throw("unreachable")
    }
}

// 第二部分，block=false且c为空
if ！block && empty(c) {
    ......
}

// 加锁，返回时释放锁
lock(&c.lock)
// 第三部分，c已经被close，且chan为空empty
if c.closed != 0 && c.qcount == 0 {
    unlock(&c.lock)
    if ep != nil {
        typedmemclr(c.elemtype, ep)
    }
    return true, false
}

// 第四部分，如果sendq队列中等待发送的sender
if sg := c.sendq.dequeue(); sg != nil {
    recv(c, sg, ep, func() { unlock(&c.lock) }, 3)
    return true,true
}

// 第五部分，没有等待的sender，buf中有数据
if c.qcount > 0 {
    qp := chanbuf(c, c.recvx)
    if ep != nil {
        typedmemmove(c, c.recvx)
        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
    }
    
    if !block {
        unlock(&c.lock)
        return false, false
    }
    // 第六部分，buf中没有元素，阻塞
}

func closechan(c *hchan) {
    if c == nil  { // chan为nil，panic
        panic(plainError("close of channel"))
    }
    
    lock(&c.lock)
    if c.closed != 0 { // chan已经closed，panic
        unlock(&c.lock)
        panic(plainError("close of closed channel"))
    }
    
    c.closed = 1
    
    var glist gList
    
    // 释放所有的reader
    for {
        sg := c.recvq.dequeue()
        ......
        gp := sg.g
        ......
        glist.push(gp)
    }
    
    // 释放所有的writer (它们会panic)
    for {
        sg := c.sendq.dequeue()
        ......
        gp := sg.g
        ......
        glist.push(gp)
    }
    unlock(&c.lock)
    
    for !glist.empty() {
        gp := glist.pop()
        gp.schedlink = 0
        goready(gp, 3)
    }
}