一篇文章带你吃透Go语言的Atomic和Channel--实战方法

type Config struct {
  NodeName string
  Addr     string
  Count    int32
}
func loadNewConfig() Config {
  return Config{NodeName: "北京", Addr: "10.77.95.27", Count: rand.Int31()}
}
func main() {
  var config atomic.Value
  config.Store(loadNewConfig())
  var cond = sync.NewCond(&sync.Mutex{}) // 设置新的config
  go func() {
    for {
      time.Sleep(time.Duration(5+rand.Int63n(5)) * time.Second)
      config.Store(loadNewConfig())
      cond.Broadcast() // 通知等待着配置已变更
    }
  }()
  go func() {
    for {
      cond.L.Lock()
      cond.Wait()                 // 等待变更信号
      c := config.Load().(Config) // 读取新的配置
      fmt.Printf("new config: %+v\n", c)
      cond.L.Unlock()
    }
  }()
  select {}
}

// lock-free的queue
type LKQueue struct {
  head unsafe.Pointer
  tail unsafe.Pointer
} 
// 通过链表实现，这个数据结构代表链表中的节点
type node struct {
  value interface{}
  next  unsafe.Pointer
}
func NewLKQueue() *LKQueue {
  n := unsafe.Pointer(&node{})
  return &LKQueue{head: n, tail: n}
} // 入队
func (q *LKQueue) Enqueue(v interface{}) {
  n := &node{value: v}
  for {
    tail := load(&q.tail)
    next := load(&tail.next)
    if tail == load(&q.tail) { // 尾还是尾
      if next == nil { // 还没有新数据入队
        if cas(&tail.next, next, n) { //增加到队尾
          cas(&q.tail, tail, n) //入队成功，移动尾巴指针
          return
        }
      } else { // 已有新数据加到队列后面，需要移动尾指针
        cas(&q.tail, tail, next)
      }
    }
  }
}
// 出队，没有元素则返回nil
func (q *LKQueue) Dequeue() interface{} {
  for {
    head := load(&q.head)
    tail := load(&q.tail)
    next := load(&head.next)
    if head == load(&q.head) { // head还是那个head
      if head == tail { // head和tail一样
        if next == nil { // 说明是空队列
          return nil
        } // 只是尾指针还没有调整，尝试调整它指向下一个
        cas(&q.tail, tail, next)
      } else { // 读取出队的数据
        v := next.value // 既然要出队了，头指针移动到下一个
        if cas(&q.head, head, next) {
          return v // Dequeue is done. return
        }
      }
    }
  }
}
// 将unsafe.Pointer原子加载转换成node
func load(p *unsafe.Pointer) (n *node) { return (*node)(atomic.LoadPointer(p)) }
// 封装CAS,避免直接将*node转换成unsafe.Pointer
func cas(p *unsafe.Pointer, old, new *node) (ok bool) {
  return atomic.CompareAndSwapPointer(p, unsafe.Pointer(old), unsafe.Pointer(new))
}

func main() {
  var ch1 = make(chan int, 10)
  var ch2 = make(chan int, 10)      // 创建SelectCase
  var cases = createCases(ch1, ch2) // 执行10次select
  for i := 0; i < 10; i++ {
    chosen, recv, ok := reflect.Select(cases)
    if recv.IsValid() { // recv case
      fmt.Println("recv:", cases[chosen].Dir, recv, ok)
    } else { // send case
      fmt.Println("send:", cases[chosen].Dir, ok)
    }
  }
}
func createCases(chs ...chan int) []reflect.SelectCase {
  var cases []reflect.SelectCase // 创建recv case
  for _, ch := range chs {
    cases = append(cases, reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(ch)})
  } // 创建send case
  for i, ch := range chs {
    v := reflect.ValueOf(i)
    cases = append(cases, reflect.SelectCase{Dir: reflect.SelectSend, Chan: reflect.ValueOf(ch), Send: v})
  }
  return cases
}

type Token struct{}
func newWorker(id int, ch chan Token, nextCh chan Token) {
  for {
    token := <-ch         // 取得令牌
    fmt.Println((id + 1)) // id从1开始
    time.Sleep(time.Second)
    nextCh <- token
  }
}
func main() {
  chs := []chan Token{make(chan Token), make(chan Token), make(chan Token), make(chan Token)} // 创建4个worker
  for i := 0; i < 4; i++ {
    go newWorker(i, chs[i], chs[(i+1)%4])
  }
  //首先把令牌交给第一个worker
  chs[0] <- struct{}{}
  select {}
}

func main() {
    go func() {
        ...... // 执行业务处理
    }()                                                                  // 处理CTRL+C等中断信号
    termChan := make(chan os.Signal)
    signal.Notify(termChan, syscall.SIGINT, syscall.SIGTERM) <- termChan // 执行退出之前的清理动作
    doCleanup()
    fmt.Println("优雅退出")
}

func main() {
  var closing = make(chan struct{})
  var closed = make(chan struct{})
  go func() { // 模拟业务处理
    for {
      select {
      case <-closing:
        return
      default: // ....... 业务计算
        time.Sleep(100 * time.Millisecond)
      }
    }
  }()            // 处理CTRL+C等中断信号
  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):
    fmt.Println("清理超时，不等了")
  }
  fmt.Println("优雅退出")
}
func doCleanup(closed chan struct{}) { time.Sleep((time.Minute))
  close(closed)

// 使用chan实现互斥锁
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 {
  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()
  fmt.Printf("locked v %v\n", ok)
  ok = m.TryLock()
  fmt.Printf("locked %v\n", ok)
}

func or(channels ...<-chan interface{}) <-chan interface{} { // 特殊情况，只有零个或者1个chan
  switch len(channels) {
  case 0:
    return nil
  case 1:
    return channels[0]
  }
  orDone := make(chan interface{})
  go func() {
    defer close(orDone)
    switch len(channels) {
    case 2: // 2个也是一种特殊情况
      select {
      case <-channels[0]:
      case <-channels[1]:
      }
    default: //超过两个，二分法递归处理
      m := len(channels) / 2
      select {
      case <-or(channels[:m]...):
      case <-or(channels[m:]...):
      }
    }
  }()
  return orDone
}
func sig(after time.Duration) <-chan interface{} {
  c := make(chan interface{})
  go func() {
    defer close(c)
    time.Sleep(after)
  }()
  return c
}
func main() {
  start := time.Now()
  <-or(sig(10*time.Second), sig(20*time.Second), sig(30*time.Second), sig(40*time.Second), sig(50*time.Second), sig(01*time.Minute))
  fmt.Printf("done after %v", time.Since(start))
}

func fanInReflect(chans ...<-chan interface{}) <-chan interface{} {
  out := make(chan interface{})
  go func() {
    defer close(out) // 构造SelectCase slice
    var cases []reflect.SelectCase
    for _, c := range chans {
      cases = append(cases, reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(c)})
    } // 循环，从cases中选择一个可用的
    for len(cases) > 0 {
      i, v, ok := reflect.Select(cases)
      if !ok { // 此channel已经close
        cases = append(cases[:i], cases[i+1:]...)
        continue
      }
      out <- v.Interface()
    }
  }()
  return out
}

func fanOut(ch <-chan interface{}, out []chan interface{}, async bool) {
  go func() {
    defer func() { //退出时关闭所有的输出chan 
      for i := 0; i < len(out); i++ {
        close(out[i])
      }
    }()
    for v := range ch { // 从输入chan中读取数据
      v := v
      for i := 0; i < len(out); i++ {
        i := i
        if async { //异步
          go func() {
            out[i] <- v // 放入到输出chan中,异步方式
          }()
        } else {
          out[i] <- v // 放入到输出chan中，同步方式
        }
      }
    }
  }()
}

func asStream(done <-chan struct{}, values ...interface{}) <-chan interface{} {
    s := make(chan interface{}) //创建一个unbuffered的channel
    go func() {                 // 启动一个goroutine，往s中塞数据
        defer close(s)             // 退出时关闭chan
        for _, v := range values { // 遍历数组
            select {
                case <-done:
                    return
                case s <- v: // 将数组元素塞入到chan中
            }
        }
    }()
    return s
}

func takeN(done <-chan struct{}, valueStream <-chan interface{}, num int) <-chan interface{} {
  takeStream := make(chan interface{}) // 创建输出流
  go func() {
    defer close(takeStream)
    for i := 0; i < num; i++ { // 只读取前num个元素
      select {
      case <-done:
        return
      case takeStream <- <-valueStream: //从输入流中读取元素
      }
    }
  }()
  return takeStream
}

func mapChan(in <-chan interface{}, fn func(interface{}) interface{}) <-chan interface{} {
  out := make(chan interface{}) //创建一个输出chan
  if in == nil {                // 异常检查
    close(out)
    return out
  }
  go func() { // 启动一个goroutine,实现map的主要逻辑
    defer close(out)
    for v := range in { // 从输入chan读取数据，执行业务操作，也就是map操作
      out <- fn(v)
    }
  }()
  return out
}
func reduce(in <-chan interface{}, fn func(r, v interface{}) interface{}) interface{} {
  if in == nil { // 异常检查
    return nil
  }
  out := <-in         // 先读取第一个元素
  for v := range in { // 实现reduce的主要逻辑
    out = fn(out, v)
  }
  return out
}
// 生成一个数据流
func asStream(done <-chan struct{}) <-chan interface{} {
  s := make(chan interface{})
  values := []int{1, 2, 3, 4, 5}
  go func() {
    defer close(s)
    for _, v := range values { // 从数组生成
      select {
      case <-done:
        return
      case s <- v:
      }
    }
  }()
  return s
}
func main() {
  in := asStream(nil)                                              // map操作: 乘以10
  mapFn := func(v interface{}) interface{} { return v.(int) * 10 } // reduce操作: 对map的结果进行累加
  reduceFn := func(r, v interface{}) interface{} { return r.(int) + v.(int) }
  sum := reduce(mapChan(in, mapFn), reduceFn) //返回累加结果
  fmt.Println(sum)
}