chan

package main

import (
   "fmt"
   "math"
   "sync"
   "sync/atomic"
   "time"
)

type semaphore chan byte
func wait(s semaphore) bool {
   _, ok := <- s
   return ok
}

func wake(s semaphore)  {
   s <- 'k'
}

type channel struct {
   sync.Mutex
   rc, wc uint32
   r, w semaphore
   i, j uint32
   size uint32
   q  []interface{}
}

func NewChannel(size uint32) *channel {
   c := new(channel)
   c.r = make(semaphore, size)
   c.w = make(semaphore, size)
   size++
   c.size = size
   c.q = make([]interface{}, size)
   return c
}

func (c *channel) Write(s interface{})  {
   for {
      c.Lock()
      j := atomic.LoadUint32(&c.j)
      i := atomic.LoadUint32(&c.i)
      if (j + 1) % c.size == i {
         atomic.AddUint32(&c.wc, 1)
         c.Unlock()
         wait(c.w)
      } else {
         j++
         j %= c.size
         c.q[j] = s
         atomic.StoreUint32(&c.j, j)
         c.Unlock()
         if atomic.LoadUint32(&c.rc) > 0 {
            atomic.AddUint32(&c.rc, math.MaxUint32)
            wake(c.r)
         }
         return
      }
   }
}

func (c *channel) Read() interface{} {
   for {
      c.Lock()
      j := atomic.LoadUint32(&c.j)
      i := atomic.LoadUint32(&c.i)
      if j == i {
         atomic.AddUint32(&c.rc, 1)
         c.Unlock()
         wait(c.r)
      } else {
         i++
         i %= c.size
         r := c.q[i]
         atomic.StoreUint32(&c.i, i)
         c.Unlock()
         if atomic.LoadUint32(&c.wc) > 0 {
            atomic.AddUint32(&c.wc, math.MaxUint32)
            wake(c.w)
         }
         return r
      }
   }
}

func (c *channel) Close() error {
   return nil
}


func main() {
   c := NewChannel(1)
   go func() {
      for i := 0; i < 700; i++ {
         //if i % 100 == 0 {
         // time.Sleep(time.Second)
         //}
         c.Write(i)
      }
      time.Sleep(time.Second*4)
      fmt.Println("end")
   }()
   for i := 0; i < 80; i++ {
      go func() {
         for
         {
            //time.Sleep(time.Minute)
            i := c.Read().(int)
            fmt.Println(i)
         }
      }()
   }
   time.Sleep(time.Minute*10)
}

标签：time,chan,atomic,func,semaphore,channel,size
来源： https://www.cnblogs.com/Janly/p/16525931.html