goroutine 调度原理 go

原文转载于 https://www.cnblogs.com/wdliu/p/9272220.html
go routine 调度
一、goroutine 简介 goroutine 本质是协程，并行计算的核心。goroutine使用方式非常简单，只需要go关键字即可启动一个协程，并且它是处于异步方式执行，并不需要等他运行完成以后在执行以后的代码

go func() { .... }() //通过go关键字启动一个协程来运行函数

二、 go routine 内部原理概念介绍在进行原理之前，先了解关键术语的概念
并发一个cpu上面能同时执行多项任务，在短时间内，cpu来回切换任务执行（在短时间内执行a，然后又迅速的切换到b执行），宏观上是同时的，微观上是顺序执行的，看起来像是多个任务同时执行，这就是并发。
并行系统有多个cpu每个cpu同一个时刻都运行任务，互不抢占自己所在的cpu的资源，同时进行，称为并行
进程 cpu在切换程序的时候，如果不保存上一个程序的状态，（我们所说的context–上下文），直接切换下一个程序，就会丢失上一个程序的一系列状态，于是引入了进程这个概念，用以划分好程序运行时所需要的资源。因此进程就是一个程序运行时侯所需要的基本资源单位（程序运行时的一个实体）。-----运行时保存有一系列资源的可执行程序实体
线程 cpu切换多个进程的时候，会花费不少时间，因为切换进程需要切换到内核态，而每次调度需要内核态都读取用户态的数据，进程一旦多起来，cpu调度会消耗一大堆资源，因此引入了线程的概念，线程本身几乎不占有资源，他们共享进程里面的资源，内核调度起来不会那么像进程切换那么消耗资源。
协程协程拥有自己的寄存器上下文和栈。协程调度切换时，将寄存器上下文和栈保存到其他地方，在切回来的时候，恢复先前保存的寄存器上下文和栈。因此，协程能保留上一次调用时的状态（即所有局部状态的一个特定组合），每次过程重入时，就相当于进入上一次调用的状态，换种说法：进入上一次离开时所处逻辑流的位置。线程和进程的操作是由程序触发系统接口，最后的执行者是系统；协程的操作执行者是用户自身程序，goroutine是协程。
调度模型简介 goroutine 能够拥有强大的并发是通过gpm调度模型实现。
M P G go的调度器内部有四个重要的结构：M，P，S，Sched
M: M代表内核级的线程，一个M是一个线程，goroutine是跑在M之上的；M是一个很大的结构，里面维护小对象内存cache（mcache），当执行goroutine，随机数发生器等等非常多的信息。
G：代表一个goroutine ，他有自己的栈，instruction pointer 和其他信息（正在等待的channel等等），用于调度。
P：P全称是Processor处理器，他的主要用途是用来执行goroutine的，他维护了一个goroutine队列，里面存储了所有需要它来执行的goroutine
Sched ：代表调度器，他维护有存储M和G的队列以及调度器的一些状态信息等。
调度实现

文章图片

从上图中看，有两个物理线程M，每一个M都拥有一个处理器P，每一个都有一个正在运行的goroutine。
P的数量可以通过 GMAXPROCS()来设置，它其实也就代表了真正的并发度，即有多少个goroutine可以同时运行。
图中灰色的goroutine并没有运行，而是处于ready的就绪态，正在等待被调度。P维护着这个队列。（称之为runqueue）
go语言里，启动一个goroutine很容易：go func(){…}()就行，所以每有一个go语句被执行，runqueue就在队列末尾加入一个
在下一个调度点，goroutine 从runqueue中取出来（现在可以认为时随机取出来）并执行。
当一个os线程MO陷入阻塞时（如下图），P转而运行M1,图中M1可能正是被创建，或者从线程缓存取出。

文章图片

当MO返回时候，他必须尝试取得一个P来运行goroutine，一般情况下，它会从其他的os线程那里拿一个P过来。
如果没有拿到的话，就把goroutine放在一个global runqueue里面，然后自己睡眠（放入线程缓存里面）。所有的P也会周期性的检查global runqueue 并运行其中的goroutine，否则global runqueue上的goroutine永远无法执行
另一种情况是p所分配的任务G很快就执行完了。（分配不均）这就导致了这个处理器p很忙，但是其他的p还有任务，此时如果global runqueue 没有任务G了，那么P不得不从其他的P那里拿来一些G来执行。一般来说，如果P从其他的P那里要拿任务的化，一般就拿runqueue 的一半，确保每个os线程都能充分的使用。如下图

文章图片

三、使用goroutine 基本使用设置goroutine运行的cpu数量，最新版本的go已经默认设置了

num := runtime.NumCPU() //获取主机的逻辑cpu的个数 runtime.GOMAXPROCS(num) //设置可同时执行的最大cpu个数

使用实例

package main import( "fmt" "time" ) func cal(a int,b int){ c:=a+b fmt.Printf("%d + %d = %d \n",a,b,c) } func main(){ for i:=0; i<10; i++{ go cal(i,i+1) //启动10个goroutine来计算 } time.Sleep(time.Second*2) //sleep是为了等待所有任务完成 }

goroutine 异常捕捉当启动多个goroutine时，如果其中一个goroutine异常了，并且我们没有进行异常处理，那么整个程序都会终，所以我们在编写程序的时候最好每个roroutine所运行的函数都进行异常处理，异常处理采用recover

package main import( "fmt" "time" ) func addele(a []int ,i int){ defer func(){ //匿名函数捕获错误 err:=recover() if err!=nil{ fmt.Println("add ele fail") } }() a[i]=i fmt.Println(a) }func main(){ Array:=make([]int a) for i:=0; i<10; i++{ go addele(Array,i) } time.Sleep(time.Second*2) } //结果 add ele fail [0 0 0 0] [0 1 0 0] [0 1 2 0] [0 1 2 3] add ele fail add ele fail add ele fail add ele fail add ele fail

同步的goroutine 由于goroutine是异步执行的，那很有可能出现主程序退出时还有goroutine没有执行完，此时goroutine也会跟着退出。如果想等到所有的 goroutine都执行完毕之后再退出程序，go提供了sync包和channel来解决同步问题，当然如果你能预测每个goroutine的执行时间，你还可以通过time.Sleep方式等待所有的goroutine执行完成以后在退出程序（如上面的例子）
示例一：使用sycn包同步goroutine
sync大致实现方式
WaitGroup等待一组goroutine执行完毕，主程序调用Add添加等待的goroutine的数量，每个goroutine的执行在结束时调用Done，此时等待队列数量减1，主程序通过Wait阻塞，直到等待队列为0

package main import ( "fmt" "sync" ) func cal(a int,b int,n* sync.WaitGroup){ c:=a+b fmt.Printf("%d + %d = %d\n",a,b,c) defer n.Done() //goroutine完成后，WaitGroup的计数-1 }func main(){ var go_sync sync.WaitGroup //声明一个WaitGroup变量 for i:=0; i<10; i++{ go_sync.Add(1) //WaitGroup 的计数加1 go cal(i,i+1,&go_sync) } go_sync.Wait() //等待所有的goroutine执行完毕 }

示例二：通过channel实现goroutine之间的同步
实现方式：通过channel能在多个goroutine之间通讯，当一个goroutine完成时候向channel发送退出信号，等待所有的 goroutine退出的时候，利用for循环 channel取channel中的信号，若取不到数据就会阻塞的原理，等待所有的goroutine执行完毕，使用该方法有个前提是你已经知道了启动了多少个goroutine

package main import( "fmt" "time" ) func cal(a int ,b int,Exitchan chan bool){ c:=a+b fmt.Printf("%d + %d =%d \n",a,b,c) time.Sleep(time.Second*2) Exitchan <-true } func main(){ Exitchan :=make(chan bool,10) for i:=0; i<10; i++{ go cal(i,i+1,Exitchan) } for j:=0; j<10; j++{ <-Exitchan //取信号，如果取不到会阻塞 } close(Exitchan)//关闭管道 }

goroutine之间的通讯 goroutine本质上是协程，可以理解为不受内核调度，而受go调度器管理的线程。goroutine之间可以通过channel进行通讯，或者实现数据共享，当然也可以使用全局变量来进行数据共享。
示例：使用channel模拟生产者和消费者模式

package main import( "fmt" "sync" ) func Productor(mychan chan int ,data int ,wait *sync.WaitGroup){ mychan<-data fmt.Println("product data:",data) wait.Done() } func Consumer(mychan chan int,wait *sync.WaitGroup){ a:=<-mychan fmt.Println("consumer data",a) wait.Done() }func main(){ datachan := make(chan int,100) //通讯数据管道 var wg sync.WaitGroup fori:=0; i<10; i++{ go Productor(datachan,i,&wg) //生产数据 wg.Add(1) } for j:=0; j<10; j++{ go Consumer(datachan,&wg) //消费数据 wg.Add(1) } wg.Wait() }

四、channel 简介 channel俗称管道，用于数据传递或数据共享，其本质是一个先进先出队列，使用goroutine+channel进行数据通讯简单高效，同时也线程安全，多个goroutine可同时修改一个channel，不需要加锁。
chnnel 可分为三种模型：
只读channel：只能读里面的数据不可以写入
只写channel：只能写数据，不可读
一般channel：可读可写
channel使用 【goroutine 调度原理】定义和声明

var readOnlyChan <-chan int//只读channel var writeOnlyChan char<- int //只写channel var mychan chan int //读写channel //定义完成以后需要 make来分配内存空间，不然使用会dedlock machannel = make(chan int,10) //或者 read_only := make (<-chan int,10)//定义只读的channel write_only := make (chan<- int,10)//定义只写的channel read_write := make (chan int,10)//可同时读写

读写数据
需要注意的是：

管道如果未关闭，在读取超时则会发生deadlock异常
管道如果关闭写入数据会pcnic
当管道中没有数据时候再行读取或读取到默认值，如int类型默认值是0

ch<- "wd" //写数据 a:=<-ch //读取数据 a,ok :=<-ch //优雅的读取数据

循环管道
需要注意的是：
- 使用range循环管道，如果管道未关闭会引发deadlock错误
- 如果采用for死循环已经关闭的管道，当管道没有数据的时候读取的数据会是管道的默认值，并且循环不会退出。

packagemain import( "fmt" "time" ) func main(){ mychannel :=make(chan int,10) for i:=0; i<10; i++{ mychannel<- i } close(mychannel)//关闭管道 fmt.Println("data length:",len(mychannel)) for v:=range mychannel{ fmt.Println(v) } fmt.Printf("data length: %d",len(mychannel)) }

带缓冲区和不带有缓冲区带缓冲区channel：定义声明时候制定了缓冲区大小（长度），可以保存多个数据。
不带缓冲区：只能存一个数据，并且只有当该数据被读取出来的时候才能存下一个数据

ch:=make(chan int) //不带缓冲区 ch:=make(chan int,10) //带缓冲去

不带缓冲区示例：

package main import "fmt" func test(c chan int){ for i:=0; i<10; i++{ fmt.Println("send",i) c<- i } } func main(){ ch:=make(chan int) go test(ch) for j:=0; j<10; j++{ fmt.Println("get",<-ch) } }

channel 实现作业池我们创建三个channel ，一个channel用于接受任务，一个channel用于保持结果，还有一个channel用于决定程序退出的时候。

package main import( "fmt" ) func Storesult(task,res chan int,exitch chan bool){ defer func(){ err:=recover() if err!=nil{ fmt.Println("do task error",err) return } }() for t:=range task{ //处理任务 fmt.Println("do task:",t) res<-t } exitch<- true //处理完发送退出信号 } func main(){ task :=maek(chan int ,20)//任务管道 res :=make(chan int ,20)//结果管道 exitchan :=make(chan bool,5) go func(){ for i:=0; i<10; i++{ task<-i } close(task) }() for i:=0; i<5; i++{ //启动五个goroutine做任务 go Storesult(task,res,exitchan) } go func(){//等5个goroutine结果 for i:=0; i<5; i++{ <-exitch } close(resch)//任务处理完成关闭结果管道，不然range报错 close(exitch)//关闭退出管道 } for re:=range res{ fmt.Println("task res:",re) } }

只读channel和只写channel 一般定义只读和只写管道意义不大，跟多时候我们可以在参数传递的时候指明管道可读还是可写，即使当前管道是可读写的。

package mainimport ( "fmt" "time" )//只能向chan里写数据 func send(c chan<- int) { for i := 0; i < 10; i++ { c <- i } } //只能取channel中的数据 func get(c <-chan int) { for i := range c { fmt.Println(i) } } func main() { c := make(chan int) go send(c) go get(c) time.Sleep(time.Second*1) }

select-case 实现非阻塞channel 原理通过select+case 加入一组管道，当满足select中的某个case的时候，那么case返回，若都不满足case，则走default分支。

package mainimport ( "fmt" )func send(c chan int){ for i :=1 ; i<10 ; i++{ c <-i fmt.Println("send data : ",i) } }func main() { resch := make(chan int,20) strch := make(chan string,10) go send(resch) strch <- "wd" select { case a := <-resch: fmt.Println("get data : ", a) case b := <-strch: fmt.Println("get data : ", b) default: fmt.Println("no channel actvie")}}

channel 频率控制在对channel进行读写的时候，go还提供了非常人性化的操作，就是对读写的频率控制，通过time.Ticke实现

package mainimport ( "time" "fmt" )func main(){ requests:= make(chan int ,5) for i:=1; i<5; i++{ requests<-i } close(requests) limiter := time.Tick(time.Second*1) for req:=range requests{ <-limiter fmt.Println("requets",req,time.Now()) //执行到这里，需要隔1秒才继续往下执行，time.Tick(timer)上面已定义 } }