Golang 基础：原生并发 goroutine channel 和 select 常见使用场景

//src/runtime/runtime2.go
type g struct {
    stack      stack   // offset known to runtime/cgo
    sched      gobuf
    goid       int64
    gopc       uintptr // pc of go statement that created this goroutine
    startpc    uintptr // pc of goroutine function
    ... ...
}

type p struct {
    lock mutex

    id          int32
    status      uint32 // one of pidle/prunning/...
  
    mcache      *mcache
    racectx     uintptr

    // Queue of runnable goroutines. Accessed without lock.
    runqhead uint32
    runqtail uint32
    runq     [256]guintptr

    runnext guintptr

    // Available G's (status == Gdead)
    gfree    *g
    gfreecnt int32

    ... ...
}



type m struct {
    g0            *g     // goroutine with scheduling stack
    mstartfn      func()
    curg          *g     // current running goroutine
    ... ...
}

// $GOROOT/src/runtime/proc.go


// forcePreemptNS is the time slice given to a G before it is
// preempted.
const forcePreemptNS = 10 * 1000 * 1000 // 10ms

c := make(chan int) //c 是一个 int 类型的 channel

func spawn(f func() error) <-chan error {
    c := make(chan error)

    go func() {
        c <- f()
    }()

    return c
}

func main() {
    c := spawn(func() error {
        time.Sleep(2 * time.Second)
        return errors.New("timeout")
    })
    fmt.Println(<-c)
}

func testNoBufferChannel() {
    var c chan int = make(chan int) //无缓冲，同步进行，没有对接人，就会阻塞住
    //var c chan int = make(chan int, 5)    //有缓冲，容量为 5

    //大多数时候，读写要在不同 goroutine，尤其是无缓冲 channel
    go func() {
        fmt.Println("goroutine run")
        b := <-c //读取 channel
        fmt.Println("read from channel: ", b)
    }()

    fmt.Println("main goroutine before write")
    c <- 1  //没有 buffer，写入 channel 时会阻塞，直到有读取
    fmt.Println("main goroutine finish")
}

main goroutine before write
goroutine run
read from channel:  1
main goroutine finish

func testBufferChannel() {
    c := make(chan int, 1)  //有缓冲，容量为 5

    //大多数时候，读写要在不同 goroutine，尤其是无缓冲 channel
    go func() {
        fmt.Println("child_goroutine run")
        b := <-c //读取 channel，有数据时不会阻塞
        fmt.Println("child_goroutine read from channel: ", b)
    }()

    fmt.Println("main goroutine before write first")
    c <- 1  //有 buffer，写入 channel 时不会阻塞，除非满了
    fmt.Println("main_goroutine first write finish, len:", len(c))

    fmt.Println("main_goroutine write second:")
    c <-2
    fmt.Println("main_goroutine finish, len:", len(c))

    time.Sleep( 3 * time.Second)    //不加这个子 goroutine 没执行就退出了
}

main goroutine before write first
main_goroutine first write finish, len: 1
main_goroutine write second:
child_goroutine run
child_goroutine read from channel:  1
main_goroutine finish, len: 1

func testSingleDirectionChannel() {

    f := func(a chan<- int, b <- chan int) {    //a 是只能写入，b 是只能读取
        x := <- a   //编译报错：Invalid operation: <- a (receive from send-only type chan<- int)
        b <- 2      //编译报错：nvalid operation: b <- 2 (send to receive-only type <-chan int)
    }
}

func testCloseChannel() {
    a := make(chan int)
    close(a)    //先关闭，然后看下几种读取关闭 channel 的结果
    b := <- a
    fmt.Println("关闭后直接读取:", b)  //0
    c, ok := <-a
    fmt.Println("关闭后通过逗号 ok 读取：", c, ok)    //0 false

    for v := range a{   //关闭的话直接跳过
        fmt.Println("关闭后通过 for-range 读取", v)
    }
}

type signal struct{}

//接收一个函数，在子 routine 里执行，然后返回一个 channel，用于主 routine 等待
func spawn(f func()) <-chan signal {
    c := make(chan signal)
    go func() {
        fmt.Println("exec f in child_routine");
        f();
        fmt.Println("f exec finished, write to channel")
        c<- signal{}
    }()
    return c
}

//测试使用无 buffer channel 实现信号
func testUseNonBufferChannelImplSignal() {
    //模拟主 routine 等待子 routine

    worker := func() {
        fmt.Println("do some work")
        time.Sleep(3 * time.Second)
    }

    fmt.Println("start a worker...")
    c := spawn(worker)

    fmt.Println("spawn finished, read channel...")
    <-c //读取，阻塞等待

    fmt.Println("worker finished")
}

start a worker...
spawn finished, read channel...
exec f in child_routine
do some work
f exec finished, write to channel
worker finished

var waitGroup sync.WaitGroup

func spawnGroup(f func(ind int), count int, groupSignal chan struct{}) <-chan signal {
	c := make(chan signal)	//用于让主 routine 阻塞的 channel
	waitGroup.Add(count)	//等待总数

	//创建 n 个 goroutine
	for i := 0; i < count; i++ {
		go func(index int) {
			<- groupSignal	//读取阻塞，等待通知执行

			//fmt.Println("exec f in child_routine, index: ", i);
			//⚠️注意上面注释的代码，这里不能直接访问 for 循环的 i，因为这个是复用的，会导致访问的值不是目标值

			fmt.Println("exec f in child_routine, index: ", index);
			f(index);
			fmt.Println(index , " exec finished, write to channel")

			waitGroup.Done()
		}(i + 1)
	}

	//创建通知主 routine 结束的 routine，不能阻塞当前函数
	go func() {
		//需要同步等待所有子 routine 执行完
		waitGroup.Wait()
		c <- signal{}	//写入数据
	}()
	return c
}


func testUseNonBufferChannelImplGroupSignal() {
	worker := func(i int) {
		fmt.Println("do some work, index ", i)
		time.Sleep(3 * time.Second)
	}

	groupSignal := make(chan struct{})
	c := spawnGroup(worker, 5, groupSignal)

	fmt.Println("main routine: close channel")
	close(groupSignal)	//通知刚创建的所有 routine


	fmt.Println("main routine: read channel...")
	<- c	//阻塞在这里

	fmt.Println("main routine: all worker finished")
}

main routine: close channel
main routine: read channel
exec f in child_routine, index:  2
do some work, index  2
exec f in child_routine, index:  1
do some work, index  1
exec f in child_routine, index:  3
do some work, index  3
exec f in child_routine, index:  4
do some work, index  4
exec f in child_routine, index:  5
do some work, index  5
4  exec finished, write to channel
5  exec finished, write to channel
3  exec finished, write to channel
1  exec finished, write to channel
2  exec finished, write to channel
main routine: all worker finished

type NewCounter struct {
	c chan int
	i int
}

func CreateNewCounter() *NewCounter {
	counter := &NewCounter{
		c: make(chan int),
		i: 0,
	}

	go func() {
		for {
			counter.i ++
			counter.c <- counter.i		//每次加一，阻塞在这里
		}
	}()

	return counter
}

func (c *NewCounter)Increase() int {
	return <- c.c		//读取到的值，是上一次加一
}

//多协程并发增加计数，通过 channel 写入阻塞，读取时加一
func testCounterWithChannel() {
	fmt.Println("\ntestCounterWithChannel ->>>")

	group := sync.WaitGroup{}
	counter := CreateNewCounter()

	for i:=0 ; i<10 ; i++ {
		group.Add(1)

		go func(i int) {
			count := counter.Increase()
			fmt.Printf("Goroutine-%d, count %d \n", i, count)
		}(i)
	}

	group.Wait()

}

var active = make(chan struct{}, 3)	//"信号量"，最多 3 个
var jobs = make(chan int, 10)

//使用带缓存的 channel，容量就是信号量的大小
func testSemaphoreWithBufferChannel() {

	//先写入数据，用作表示任务
	go func() {
		for i:= 0; i < 9; i++ {
			jobs <- i + 1
		}
		close(jobs)
	}()

	var wg sync.WaitGroup

	for j := range jobs {
		wg.Add(1)

		//执行任务
		go func(i int) {
			//通知开始执行，当容量用完时，阻塞
			active <- struct{}{}

			//fmt.Println("exec job ", i)
			log.Printf("exec job: %d, length of active: %d \n", i, len(active))
			time.Sleep(2 * time.Second)

			//执行完，通知结束
			<- active
			wg.Done()

		}(j)
	}

	wg.Wait()
}

2022/04/20 19:14:26 exec job: 1, length of active: 1 
2022/04/20 19:14:26 exec job: 9, length of active: 2 
2022/04/20 19:14:26 exec job: 5, length of active: 3 
2022/04/20 19:14:28 exec job: 6, length of active: 3 
2022/04/20 19:14:28 exec job: 7, length of active: 3 
2022/04/20 19:14:28 exec job: 8, length of active: 3 
2022/04/20 19:14:30 exec job: 3, length of active: 3 
2022/04/20 19:14:30 exec job: 2, length of active: 3 
2022/04/20 19:14:30 exec job: 4, length of active: 3 

func testSelect() {
    channelA := make(chan int)
    channelB := make(chan int)

    go func() {
        var readA bool
        var readB bool

        for {
            select {
            case x := <- channelA:
                fmt.Println("child_routine: read from channelA:", x)
                readA = true
            case y := <- channelB:
                fmt.Println("child_routine:  read from channelB:", y)
                readB = true
            //default:
            //  //其他 case 阻塞，就执行 default
            //  fmt.Println("default")
            }

            if readA && readB {

                fmt.Println("child_goroutine finish")
                return;
            } else {
                fmt.Println("child_goroutine still loop, ", readA, readB)
            }
        }
    }()

    fmt.Println("main goroutine")

    time.Sleep(2 * time.Second)

    fmt.Println("main goroutine, write to channelA")
    channelA <- 111
    fmt.Println("main goroutine, write to channelA finish")

    time.Sleep(1 * time.Second)

    fmt.Println("main goroutine, write to channelB")
    channelB <- 111
    fmt.Println("main goroutine, write to channelB finish")

    time.Sleep( 5 * time.Second)
    fmt.Println("main goroutinefinish")
}

main goroutine
main goroutine, write to channelA
main goroutine, write to channelA finish
child_routine: read from channelA: 111
child_goroutine still loop,  true false
main goroutine, write to channelB
main goroutine, write to channelB finish
child_routine:  read from channelB: 111
child_goroutine finish
main goroutinefinish

func tryWriteChannel(c chan<- int, value int) bool {
	select {
	case c <- value
		return true
	default:	//其他没就绪时，会执行
		return false
	}
}

			//active <- 1		//之前直接写 channel，如果满了，就会阻塞
			writed := tryWriteChannel(active, 1)	//改成这样，可以在阻塞时，处理相关逻辑
			if !writed {
				log.Println("failed to write channel")
				return
			}

func tryGetSemaphore(c chan<- struct{}) bool {
	select {
	case c <- struct {}{}:
		return true
	case <- time.After(1 * time.Second):		//在写 channel 的基础上，额外加一个情况，超时情况
		log.Println("timeout!!!")
		//1s 后返回，可以在这里做超时处理
		return true
	}
}

func worker() {
  heartbeat := time.NewTicker(30 * time.Second)
  defer heartbeat.Stop()
  for {
    select {
    case <-c:
      // ... do some stuff
    case <- heartbeat.C:
      //... do heartbeat stuff
    }
  }
}