select

select

Golangselect就是监听IO操作,当IO操作发生时,触发相应的动作。在执行select语句的时候,运行时系统会自上而下地判断每个case中的发送或接收操作是否可以被立即执行(当前Goroutine不会因此操作而被阻塞)select的用法与switch非常类似,由select开始一个新的选择块,每个选择条件由case语句来描述。

var c1, c2 <-chan interface{}
var c3 chan<- interface{}
select {
	case <-c1:
		// Do something
	case <-c2:
		// Do something
	case c3 <- struct{}{}:
		// Do something
}

switch相同的是,select代码块也包含一系列case分支。跟switch不同的是,case分支不会被顺序测试,如果没有任何分支的条件可供满足,select会一直等待直到某个case语句完成。所有通道的读取和写入都被同时考虑,以查看它们中的任何一个是否准备好:如果没有任何通道准备就绪,则整个select语句将会阻塞。当一个通道准备好时,该操作将继续,并执行相应的语句。我们来看一个简单的例子:

start := time.Now()
c := make(chan interface{})
go func() {
	time.Sleep(5 * time.Second)
	close(c) // 5 秒后关闭通道
}()

fmt.Println("Blocking on read...")
select {
case <-c: // 尝试读取通道。注意,尽管我们可以不使用select语句而直接使用<-c,但我们的目的是为了展示select语句。
	fmt.Printf("Unblocked %v later.\n", time.Since(start))
}

// Blocking on read...
// Unblocked 5s later.

通道是将Goroutine的粘合剂,select语句是通道的粘合剂。后者让我们能够在项目中组合通道以形成更大的抽象来解决实际中遇到的问题。我们可以在单个函数或类型定义中找到将本地通道绑定在一起的select语句,也可以在全局范围找到连接系统级别两个或多个组件的使用范例。除了连接组件外,在程序中的关键部分,select语句还可以帮助你安全地将通道与业务层面的概念(如取消,超时,等待和默认值)结合在一起。

select顺序

对于空的select语句,程序会被阻塞,准确的说是当前协程被阻塞,同时Golang自带死锁检测机制,当发现当前协程再也没有机会被唤醒时,则会panic。所以上述程序会panic

package main

func main() {
    select {
    }
}

select中各个case执行顺序是随机的,如果某个case中的channel已经ready,则执行相应的语句并退出select流程,如果所有case中的channel都未ready,则执行default中的语句然后退出select流程。另外,由于启动的协程和select语句并不能保证执行顺序,所以也有可能select执行时协程还未向channel中写入数据,所以select直接执行default语句并退出。所以,以下三种输出都有可能:

package main

import (
    "fmt"
    "time"
)

func main() {
    chan1 := make(chan int)
    chan2 := make(chan int)

    go func() {
        chan1 <- 1
        time.Sleep(5 * time.Second)
    }()

    go func() {
        chan2 <- 1
        time.Sleep(5 * time.Second)
    }()

    select {
    case <-chan1:
        fmt.Println("chan1 ready.")
    case <-chan2:
        fmt.Println("chan2 ready.")
    default:
        fmt.Println("default")
    }

    fmt.Println("main exit.")
}

select会按照随机的顺序检测各case语句中channel是否ready,如果某个case中的channel已经ready则执行相应的case语句然后退出select流程,如果所有的channel都未ready且没有default的话,则会阻塞等待各个channel。所以上述程序会一直阻塞。

package main

import (
    "fmt"
    "time"
)

func main() {
    chan1 := make(chan int)
    chan2 := make(chan int)

    writeFlag := false
    go func() {
        for {
            if writeFlag {
                chan1 <- 1
            }
            time.Sleep(time.Second)
        }
    }()

    go func() {
        for {
            if writeFlag {
                chan2 <- 1
            }
            time.Sleep(time.Second)
        }
    }()

    select {
    case <-chan1:
        fmt.Println("chan1 ready.")
    case <-chan2:
        fmt.Println("chan2 ready.")
    }

    fmt.Println("main exit.")
}

程序中声明两个channel,分别为chan1chan2,依次启动两个协程,协程会判断一个bool类型的变量writeFlag来决定是否要向channel中写入数据,由于writeFlag永远为false,所以实际上协程什么也没做。select语句两个case分别检测chan1chan2是否可读,这个select语句不包含default语句。

总结而言:

  • select语句中除default外,每个case操作一个channel,要么读要么写
  • select语句中除default外,各case执行顺序是随机的
  • select语句中如果没有default语句,则会阻塞等待任一case
  • select语句中读操作要判断是否成功读取,关闭的channel也可以读取

select的典型应用

定时器

func main() {

   tickTimer := time.NewTicker(1 * time.Second)
   barTimer := time.NewTicker(60 * time.Second)
   for {
   	select {
   	case <-tickTimer.C:
   		fmt.Println("tick")
   	case <-barTimer.C:
   		fmt.Println("bar")
   	}
   }
}

select实现原理

Golang实现select时,定义了一个数据结构表示每个case语句(defautdefault实际上是一种特殊的case)select执行过程可以类比成一个函数,函数输入case数组,输出选中的case,然后程序流程转到选中的case块。

case数据结构

源码包 src/runtime/select.go:scase 定义了表示case语句的数据结构:

type scase struct {
	c           *hchan         // chan
	kind        uint16
	elem        unsafe.Pointer // data element
}

scase.c为当前case语句所操作的channel指针,这也说明了一个case语句只能操作一个channelscase.kind表示该case的类型,分为读channel、写channeldefault,三种类型分别由常量定义:

  • caseRecv:case语句中尝试读取scase.c中的数据;
  • caseSend:case语句中尝试向scase.c中写入数据;
  • caseDefault:default语句

scase.elem表示缓冲区地址,跟据scase.kind不同,有不同的用途:

  • scase.kind == caseRecv:scase.elem表示读出channel的数据存放地址;
  • scase.kind == caseSend:scase.elem表示将要写入channel的数据存放地址;

select实现逻辑

源码包src/runtime/select.go:selectgo()定义了select选择case的函数:

func selectgo(cas0 *scase, order0 *uint16, ncases int) (int, bool)

函数参数:

  • cas0scase数组的首地址,selectgo()就是从这些scase中找出一个返回。
  • order0为一个两倍cas0数组长度的buffer,保存scase随机序列pollorderscasechannel地址序列lockorder
    • pollorder:每次selectgo执行都会把scase序列打乱,以达到随机检测case的目的。
    • lockorder:所有case语句中channel序列,以达到去重防止对channel加锁时重复加锁的目的。
  • ncases表示scase数组的长度

函数返回值:

  1. int:选中case的编号,这个case编号跟代码一致
  2. bool:是否成功从channle中读取了数据,如果选中的case是从channel中读数据,则该返回值表示是否读取成功。

selectgo实现伪代码如下:

func selectgo(cas0 *scase, order0 *uint16, ncases int) (int, bool) {
    //1. 锁定scase语句中所有的channel
    //2. 按照随机顺序检测scase中的channel是否ready
    //   2.1 如果case可读,则读取channel中数据,解锁所有的channel,然后返回(case index, true)
    //   2.2 如果case可写,则将数据写入channel,解锁所有的channel,然后返回(case index, false)
    //   2.3 所有case都未ready,则解锁所有的channel,然后返回(default index, false)
    //3. 所有case都未ready,且没有default语句
    //   3.1 将当前协程加入到所有channel的等待队列
    //   3.2 当将协程转入阻塞,等待被唤醒
    //4. 唤醒后返回channel对应的case index
    //   4.1 如果是读操作,解锁所有的channel,然后返回(case index, true)
    //   4.2 如果是写操作,解锁所有的channel,然后返回(case index, false)
}

func selectgo(sel *hselect) int {
	// ...

	// case洗牌
	pollslice := slice{unsafe.Pointer(sel.pollorder), int(sel.ncase), int(sel.ncase)}
	pollorder := *(*[]uint16)(unsafe.Pointer(&pollslice))
	for i := 1; i < int(sel.ncase); i++ {
		//....
	}

	// 给case排序
	lockslice := slice{unsafe.Pointer(sel.lockorder), int(sel.ncase), int(sel.ncase)}
	lockorder := *(*[]uint16)(unsafe.Pointer(&lockslice))
	for i := 0; i < int(sel.ncase); i++ {
		// ...
	}
	for i := int(sel.ncase) - 1; i >= 0; i-- {
		// ...
	}

	// 加锁该select中所有的channel
	sellock(scases, lockorder)

	// 进入loop
loop:
	// ...
	// pass 1 - look for something already waiting
	// 按顺序遍历case来寻找可执行的case
	for i := 0; i < int(sel.ncase); i++ {
		//...
		switch cas.kind {
		case caseNil:
			continue
		case caseRecv:
			// ... goto xxx
		case caseSend:
			// ... goto xxx
		case caseDefault:
			dfli = casi
			dfl = cas
		}
	}

	// 没有找到可以执行的case,但有default条件,这个if里就会直接退出了。
	if dfl != nil {
		// ...
	}
	// ...

	// pass 2 - enqueue on all chans
	// chan入等待队列
	for _, casei := range lockorder {
		// ...
		switch cas.kind {
		case caseRecv:
			c.recvq.enqueue(sg)

		case caseSend:
			c.sendq.enqueue(sg)
		}
	}

	// wait for someone to wake us up
	// 等待被唤起,同时解锁channel(selparkcommit这里实现的)
	gp.param = nil
	gopark(selparkcommit, nil, "select", traceEvGoBlockSelect, 1)

	// 突然有故事发生,被唤醒,再次该select下全部channel加锁
	sellock(scases, lockorder)

	// pass 3 - dequeue from unsuccessful chans
	// 本轮最后一次循环操作,获取可执行case,其余全部出队列丢弃
	casi = -1
	cas = nil
	sglist = gp.waiting
	// Clear all elem before unlinking from gp.waiting.
	for sg1 := gp.waiting; sg1 != nil; sg1 = sg1.waitlink {
		sg1.isSelect = false
		sg1.elem = nil
		sg1.c = nil
	}
	gp.waiting = nil

	for _, casei := range lockorder {
		// ...
		if sg == sglist {
			// sg has already been dequeued by the G that woke us up.
			casi = int(casei)
			cas = k
		} else {
			c = k.c
			if k.kind == caseSend {
				c.sendq.dequeueSudoG(sglist)
			} else {
				c.recvq.dequeueSudoG(sglist)
			}
		}
		// ...
	}

	// 没有的话,再走一次loop
	if cas == nil {
		goto loop
	}
	// ...
bufrecv:
	// can receive from buffer
bufsend:
	// ...
recv:
	// ...
rclose:
	// ...
send:
	// ...
retc:
	// ...
sclose:
	// send on closed channel
}

特别说明:对于读channelcase来说,如case elem, ok := <-chan1:,如果channel有可能被其他协程关闭的情况下,一定要检测读取是否成功,因为closechannel也有可能返回,此时ok == false

selectgo 流程分析

上一页