go channel 原理及使用场景

type hchan struct {  qcount   uint           // Channel 中的元素个数  dataqsiz uint           // Channel 中的循环队列的长度  buf      unsafe.Pointer // Channel 的缓冲区数据指针  elemsize uint16 // 当前 Channel 能够收发的元素大小  closed   uint32  elemtype *_type // 当前 Channel 能够收发的元素类型  sendx    uint   // Channel 的发送操作处理到的位置  recvx    uint   // Channel 的接收操作处理到的位置  recvq    waitq  // 当前 Channel 由于缓冲区空间不足而阻塞的 Goroutine 列表，双向链表(sugog)  sendq    waitq  // 当前 Channel 由于缓冲区空间不足而阻塞的 Goroutine 列表，双向链表(sugog)
  // 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
  // compiler checks this but be safe.  if elem.size >= 1<<16 {    throw("makechan: invalid channel element type")  }  if hchanSize%maxAlign != 0 || elem.align > maxAlign {    throw("makechan: bad alignment")  }
  // 如果内存超了，或者分配的内存大于channel最大分配内存，或者分配的size小于0，直接Panic  mem, overflow := math.MulUintptr(elem.size, uintptr(size))  if overflow || mem > maxAlloc-hchanSize || size < 0 {    panic(plainError("makechan: size out of range"))  }
  // Hchan does not contain pointers interesting for GC when elements stored in buf do not contain pointers.  // buf points into the same allocation, elemtype is persistent.  // SudoG's are referenced from their owning thread so they can't be collected.  // TODO(dvyukov,rlh): Rethink when collector can move allocated objects.  var c *hchan  switch {  case mem == 0:    // 如果没有缓冲区，分配一段内存    c = (*hchan)(mallocgc(hchanSize, nil, true))    // Race detector uses this location for synchronization.    c.buf = c.raceaddr()  case elem.ptrdata == 0:    // 有缓冲时，如果元素不包含指针类型,会为当前的 Channel 和底层的数组分配一块连续的内存空间    c = (*hchan)(mallocgc(hchanSize+mem, nil, true))    c.buf = add(unsafe.Pointer(c), hchanSize)  default:    // 有缓冲区，且元素包含指针类型，channel和buf数组各自分配内存    c = new(hchan)    c.buf = mallocgc(mem, elem, true)  }
  // 元素大小，元素类型，循环数组长度，更新到channel  c.elemsize = uint16(elem.size)  c.elemtype = elem  c.dataqsiz = uint(size)  lockInit(&c.lock, lockRankHchan)
  if debugChan {    print("makechan: chan=", c, "; elemsize=", elem.size, "; dataqsiz=", size, "\n")  }  return c}

  func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {    lock(&c.lock)
    if c.closed != 0 {      unlock(&c.lock)      panic(plainError("send on closed channel"))    }

  // 如果目标 Channel 没有被关闭并且已经有处于读等待的 Goroutine，那么 runtime.chansend 会从接收队列 recvq 中取出最先陷入等待的 Goroutine 并直接向它发送数据  if sg := c.recvq.dequeue(); sg != nil {      send(c, sg, ep, func() { unlock(&c.lock) }, 3)      return true    }
  //   func send(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {    if sg.elem != nil {      // 调用 runtime.sendDirect 将发送的数据直接拷贝到 x = <-c 表达式中变量 x 所在的内存地址上      sendDirect(c.elemtype, sg, ep)      sg.elem = nil    }    gp := sg.g    unlockf()    gp.param = unsafe.Pointer(sg)    // 调用 runtime.goready 将等待接收数据的 Goroutine 标记成可运行状态 Grunnable 并把该 Goroutine 放到发送方所在的处理器的 runnext 上等待执行，该处理器在下一次调度时会立刻唤醒数据的接收方；    // 需要注意的是，发送数据的过程只是将接收方的 Goroutine 放到了处理器的 runnext 中，程序没有立刻执行该 Goroutine    goready(gp, skip+1)  }

  func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {    ...    // 如果当前元素数小于循环队列的长度    if c.qcount < c.dataqsiz {      // 使用 runtime.chanbuf 计算出下一个可以存储数据的位置      qp := chanbuf(c, c.sendx)      // 将发送的数据拷贝到缓冲区中      typedmemmove(c.elemtype, qp, ep)      // 发送的位置索引+1      c.sendx++      // 如果循环队列满了就从0开始      // 因为这里的 buf 是一个循环数组，所以当 sendx 等于 dataqsiz 时会重新回到数组开始的位置      if c.sendx == c.dataqsiz {        c.sendx = 0      }      // 增加当前元素数      c.qcount++      unlock(&c.lock)      return true    }    ...  }

  func chansend(c *hchan, ep unsafe.Pointer, block bool, callerpc uintptr) bool {    ...    // 使用 select 关键字可以向 Channel 非阻塞地发送消息    if !block {      unlock(&c.lock)      return false    }
    // 获取发送数据使用的 Goroutine    gp := getg()    //  获取 runtime.sudog 结构    mysg := acquireSudog()    // 设置待发送数据的内存地址    mysg.elem = ep    // 设置发送数据的goroutine    mysg.g = gp    mysg.isSelect = false    // 设置发送的channel    mysg.c = c    // 设置到goroutine的waiting上    gp.waiting = mysg    // 加入到发送等待队列    c.sendq.enqueue(mysg)    // 阻塞等待唤醒    atomic.Store8(&gp.parkingOnChan, 1)    gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanSend, traceEvGoBlockSend, 2)    KeepAlive(ep)
    // someone woke us up.    if mysg != gp.waiting {      throw("G waiting list is corrupted")    }    gp.waiting = nil    gp.activeStackChans = false    closed := !mysg.success    gp.param = nil    if mysg.releasetime > 0 {      blockevent(mysg.releasetime-t0, 2)    }    mysg.c = nil    releaseSudog(mysg)    if closed {      if c.closed == 0 {        throw("chansend: spurious wakeup")      }      panic(plainError("send on closed channel"))    }    return true  }

    if c == nil {      if !block {        return      }      // 让出使用权      gopark(nil, nil, waitReasonChanReceiveNilChan, traceEvGoStop, 2)      throw("unreachable")    }
    // 不获取锁的情况下，检查失败的非阻塞操作    if !block && empty(c) {      // 显示未关闭，继续返回false，因为channel不会重新打开      if atomic.Load(&c.closed) == 0 {        return      }
      if empty(c) {        // The channel is irreversibly closed and empty.        if raceenabled {          raceacquire(c.raceaddr())        }        // Channel 已经被关闭并且缓冲区中不存在任何数据，那么会清除 ep 指针中的数据并立刻返回        if ep != nil {          typedmemclr(c.elemtype, ep)        }        return true, false      }    }
    var t0 int64    if blockprofilerate > 0 {      t0 = cputicks()    }
    lock(&c.lock)
    if c.closed != 0 && c.qcount == 0 {      if raceenabled {        raceacquire(c.raceaddr())      }      unlock(&c.lock)      // Channel 已经被关闭并且缓冲区中不存在任何数据，那么会清除 ep 指针中的数据并立刻返回      if ep != nil {        typedmemclr(c.elemtype, ep)      }      return true, false    }

  if sg := c.sendq.dequeue(); sg != nil {      recv(c, sg, ep, func() { unlock(&c.lock) }, 3)      return true, true    }

  func recv(c *hchan, sg *sudog, ep unsafe.Pointer, unlockf func(), skip int) {    if c.dataqsiz == 0 {      if raceenabled {        racesync(c, sg)      }      if ep != nil {        // 如果无缓存，直接从发送者拷贝数据        recvDirect(c.elemtype, sg, ep)      }    } else {      // 由于队列已满，接收数据的索引和发送数据的索引一致      qp := chanbuf(c, c.recvx)      if raceenabled {        racenotify(c, c.recvx, nil)        racenotify(c, c.recvx, sg)      }      // 数据从队列拷贝到目标内存地址      if ep != nil {        typedmemmove(c.elemtype, ep, qp)      }      // 数据从发送者拷贝到缓冲区      typedmemmove(c.elemtype, qp, sg.elem)      c.recvx++      if c.recvx == c.dataqsiz {        c.recvx = 0      }      c.sendx = c.recvx // c.sendx = (c.sendx+1) % c.dataqsiz    }    sg.elem = nil    gp := sg.g    unlockf()    gp.param = unsafe.Pointer(sg)    sg.success = true    if sg.releasetime != 0 {      sg.releasetime = cputicks()    }    // 无论发生哪种情况，运行时都会调用 runtime.goready 将当前处理器的 runnext 设置成发送数据的 Goroutine，在调度器下一次调度时将阻塞的发送方唤醒。    goready(gp, skip+1)  }

  if c.qcount > 0 {      // 直接从队列取数据      qp := chanbuf(c, c.recvx)      if raceenabled {        racenotify(c, c.recvx, nil)      }      // 放到目标内存      if ep != nil {        typedmemmove(c.elemtype, ep, qp)      }      // 清空队列中对应的元素      typedmemclr(c.elemtype, qp)      // 接收索引+1      c.recvx++      if c.recvx == c.dataqsiz {        c.recvx = 0      }      // 队列元素-1      c.qcount--      unlock(&c.lock)      return true, true    }

  if !block {      unlock(&c.lock)      return false, false    }
    // no sender available: block on this channel.    gp := getg()    mysg := acquireSudog()    mysg.releasetime = 0    if t0 != 0 {      mysg.releasetime = -1    }    // No stack splits between assigning elem and enqueuing mysg    // on gp.waiting where copystack can find it.    mysg.elem = ep    mysg.waitlink = nil    gp.waiting = mysg    mysg.g = gp    mysg.isSelect = false    mysg.c = c    gp.param = nil    c.recvq.enqueue(mysg)    // 阻塞等待，让出使用权    atomic.Store8(&gp.parkingOnChan, 1)    gopark(chanparkcommit, unsafe.Pointer(&c.lock), waitReasonChanReceive, traceEvGoBlockRecv, 2)
    // 唤醒之后清空sudog    if mysg != gp.waiting {      throw("G waiting list is corrupted")    }    gp.waiting = nil    gp.activeStackChans = false    if mysg.releasetime > 0 {      blockevent(mysg.releasetime-t0, 2)    }    success := mysg.success    gp.param = nil    mysg.c = nil    releaseSudog(mysg)    return true, success

  func closechan(c *hchan) {    if c == nil {      panic(plainError("close of nil channel"))    }
    lock(&c.lock)    if c.closed != 0 {      unlock(&c.lock)      panic(plainError("close of closed channel"))    }

  c.closed = 1
    var glist gList
    // release all readers    for {      sg := c.recvq.dequeue()      if sg == nil {        break      }      if sg.elem != nil {        typedmemclr(c.elemtype, sg.elem)        sg.elem = nil      }      if sg.releasetime != 0 {        sg.releasetime = cputicks()      }      gp := sg.g      gp.param = unsafe.Pointer(sg)      sg.success = false      if raceenabled {        raceacquireg(gp, c.raceaddr())      }      glist.push(gp)    }
    // release all writers (they will panic)    for {      sg := c.sendq.dequeue()      if sg == nil {        break      }      sg.elem = nil      if sg.releasetime != 0 {        sg.releasetime = cputicks()      }      gp := sg.g      gp.param = unsafe.Pointer(sg)      sg.success = false      if raceenabled {        raceacquireg(gp, c.raceaddr())      }      glist.push(gp)    }    unlock(&c.lock)
    // 为所有被阻塞的 Goroutine 调用 runtime.goready 触发调度。    for !glist.empty() {      gp := glist.pop()      gp.schedlink = 0      goready(gp, 3)    }

package main
import (  "fmt"  "time")
func main() {  // 4个channel  chs := make([]chan int, 4)  for i, _ := range chs {    chs[i] = make(chan int)    // 开4个协程    go func(i int) {      for {        // 获取当前channel值并打印        v := <-chs[i]        fmt.Println(v + 1)        time.Sleep(time.Second)        // 把下一个值写入下一个channel，等待下一次消费        chs[(i+1)%4] <- (v + 1) % 4      }
    }(i)  }
  // 往第一个塞入0  chs[0] <- 0  select {}}

package main
import (  "fmt"  "time")
func main() {  // 每次处理3个请求  chLimit := make(chan struct{}, 3)  for i := 0; i < 20; i++ {    chLimit <- struct{}{}    go func(i int) {      fmt.Println("下游服务处理逻辑...", i)      time.Sleep(time.Second * 3)      <-chLimit    }(i)  }  time.Sleep(30 * time.Second)}

package main
import (  "fmt"  "sync"  "time")
var wg sync.WaitGroup
func main() {  // 每次处理3个请求  chLimit := make(chan struct{}, 3)  for i := 0; i < 20; i++ {    chLimit <- struct{}{}    wg.Add(1)    go func(i int) {      fmt.Println("下游服务处理逻辑...", i)      time.Sleep(time.Second * 3)      <-chLimit      wg.Done()    }(i)  }  wg.Wait()}

package main
import (  "fmt"  "log"  "os"  "os/signal"  "syscall"  "time")
func main() {  var closing = make(chan struct{})  var closed = make(chan struct{})
  go func() {    for {      select {      case <-closing:        return      default:        fmt.Println("业务逻辑...")        time.Sleep(1 * time.Second)      }    }  }()
  termChan := make(chan os.Signal)  // 监听退出信号  signal.Notify(termChan, syscall.SIGINT, syscall.SIGTERM)  <-termChan
  // 退出中  close(closing)
  // 退出之前清理一下  go doCleanup(closed)
  select {  case <-closed:  case <-time.After(time.Second):    log.Println("清理超时不等了")  }
  log.Println("优雅退出")}
func doCleanup(closed chan struct{}) {  time.Sleep(time.Minute)  // 清理完后退出  close(closed)}

package main
import (  "log"  "time")
type Mutex struct {  ch chan struct{}}
// 初始化锁func NewMutex() *Mutex {  mu := &Mutex{make(chan struct{}, 1)}  mu.ch <- struct{}{}  return mu}
// 加锁，阻塞获取func (m *Mutex) Lock()  {  <- m.ch}
// 释放锁func (m *Mutex) Unlock()  {  select {    // 成功写入channel代表释放成功  case m.ch <- struct{}{}:  default:    panic("unlock of unlocked mutex")  }}
// 尝试获取锁func (m *Mutex) TryLock() bool {  select {  case <-m.ch:    return true  default:
  }  return false}
func (m *Mutex) LockTimeout(timeout time.Duration) bool {  timer := time.NewTimer(timeout)
  select {  case <-m.ch:    // 成功获取锁关闭定时器    timer.Stop()    return true  case <-timer.C:
  }  // 获取锁超时  return false}
// 是否上锁func (m *Mutex) IsLocked() bool {  return len(m.ch) == 0}

func main()  {  m := NewMutex()  ok := m.TryLock()  log.Printf("locked v %v\n", ok)  ok = m.TryLock()  log.Printf("locked v %v\n", ok)
  go func() {    time.Sleep(5*time.Second)    m.Unlock()  }()
  ok = m.LockTimeout(10*time.Second)  log.Printf("LockTimeout v %v\n", ok)}