GMP模型 && channel

• 1. 前言
• 2. GMP模型
• • 2.1. 基本概念
  • 2.2. 调度器策略
  • 2.3. go指令的调度流程
  • 2.4. go启动周期的M0和G0
  • 2.5. GMP可视化
  • 2.6. GMP的几种调度场景
• 3. channel
• • 3.1. channel的基本使用
  • 3.2. 同步器

1. 前言

Go中的并发是函数相互独立运行的体现，Goroutines是并发运行的函数。

1. 并发：多线程程序在单核CPU上执行，每个线程使用时间片轮转执行，间隔是ms级别的
2. 并行： 多线程程序在多核CPU上执行，每个cpu上都执行一个线程，同一时刻有多个线程执行
3. go主线程是一个物理线程(内核态)，可以发起多个协程goroutine，协程是一个轻量级线程(逻辑态)
4. goroutine的特点：有独立的栈空间；共享程序堆空间；调度由用户控制
5. 创建一个协程：go 任务函数

2. GMP模型

2.1. 基本概念

M：代表内核线程，记录内核线程栈信息，当goroutine调度到线程时，使用该goroutine自己的栈信息

P：调度器processor，负责调度goroutine，维护一个本地goroutine队列，主线程从调度器上获得goroutine并执行，同时还负责部分内存的管理。

G：表示goroutine，每个goroutine都有自己的栈空间，定时器，初始化的栈空间在2k左右，空间会随着需求增长

M代表一个工作线程，在M上有一个P和G，P是绑定到M上的，G是通过P的调度获取的，在某一时刻，一个M上只有一个G（g0除外）在P上拥有一个G队列，里面是已经就绪的G，是可以被调度到线程栈上执行的协程，称为运行队列

全局队列：存放等待运行的G

P的本地队列：优先将新创建的G存入到P的本地队列，如果本地队列已满，则存入到全局队列

P列表：程序启动时创建，P的最大个数==GOMAXPROCS

M列表：当前OS分配到go程序的内核线程数

2.2. 调度器策略

1. 复用线程
   • work stealing机制：当本线程无可运行的G时，尝试从其他线程绑定的调度器中偷取协程，而不是销毁线程
   • hand off机制：当本地线程由于G发生阻塞时，线程释放绑定的P，将P转给其他空闲的M线程来执行
2. 并行： GOMAXPROCS设置P的数量，最多有GOMAXPROCS个线程分布在多个CPU上同时运行
3. 抢占 go中的协程最多只能占用CPU 10ms，防止其他协程处于饥饿状态
4. 全局G队列 当M线程执行work stealing机制从其他调度器P中获取不到G时，可以从全局协程队列中获取

2.3. go指令的调度流程

• 通过 go func() 创建一个goroutine
• 有两个存储G的队列，一个是局部调度器P的本地协程队列，一个是全局协程队列。新创建的G保存在P的本地队列中，如果P的本地队列已满则保存在全局队列中
• G只能运行在M中，一个M必须持有一个P。M从本地协程队列中选取一个可执行的G来执行，如果本地队列为空则采用stealing机制从其他协程队列中获取一个G来执行
• 一个M调度G的执行过程其实是一个循环的过程
  • 当M执行某一个G时发生syscall（系统调用）或者其他阻塞操作，M发生阻塞。如果当前一些G正在执行，runtime会将这个线程M从P中移除，然后创建一个新的内核线程M来执行G（如果有空闲的内核线程可进行复用）
  • 当M系统调用结束时，此时G会尝试获取一个空闲的P执行，放入到P的本地队列中。如果获取不到调度器P，那么线程M则进行休眠，加入到空闲线程队列中，协程G则放入到全局协程队列中

2.4. go启动周期的M0和G0

M0

• 启动程序后编号为0的主线程
• 在全局变量runtime.m0中，不需要在heap上分配
• 负责执行初始化操作和启动第一个协程G
• 启动第一个协程之后，M0与其他主线程相同

G0

• 每次启动一个M，第一个创建的goroutine即为G0，存放在全局空间
• G0仅仅用于调度，不指向任何可执行的函数
• 每个M都会绑定一个属于自己的G0
• 在syscall时会将M切换到G0，再调度

2.5. GMP可视化

• 创建trace文件
• 启动trace
• 执行业务代码
• 停止trace
• go run运行后，得到trace.out文件
• 通过go tool trace trace文件名 可视化查看GMP

func main() {f, err := os.Create("trace.out")if err != nil {panic(err)}defer f.Close()err = trace.Start(f)if err != nil {panic(err)}fmt.Println("业务逻辑...")fmt.Printf("GMP Model Trace test")trace.Stop()
}

2.6. GMP的几种调度场景

• G1创建G2：P拥有G1，M获取P开始运行G1，G1使用go func() 创建了G2，为了局部性G2优先加入到G1所在的本地协程队列

• G1执行完毕：当G1调用goexit()退出时，M切换到所绑定的协程为G0，由G0负责调度本地协程队列中的G2，交给M执行

• G2执行过程中开辟过多的协程：如果G2运行时需要创建6个协程，本地队列只能存放四个G3-G6，在创建G7时需要将本地协程队列的前两个协程与G7协程同时放入到全局协程队列中，此时本地协程队列还有一半空间，可以直接创建G8协程

• 唤醒休眠队列的线程：在创建G时，运行的G会尝试唤醒其他空闲的调度器与内核线程组合进行绑定。假定G2唤醒了线程M2，M2绑定了P2，并且运行了G0，但P2的本地协程队列没有协程（空队列），此时M2为自旋线程

• 自旋线程M2从全局协程队列中批量获取n个G。其中，GQ为全局协程队列的size，GOMAXPROCS为当前调度器个数。可以看作是从全局协程队列到本地协程队列的一种负载均衡策略

  n = min(len(GQ)/GOMAXPROCS+1, len(GQ/2))

• 如果全局协程队列为空，自旋线程M2会执行work stealing机制，从其他调度器P的本地队列中获取一半协程G到M2的本地队列

• 自旋线程个数 + 执行线程个数 ≤ GOMAXPROCS

3. channel

3.1. channel的基本使用

为什么需要channel?

1.主线程在等待所有goroutine全部完成的时间很难确定

2.通过全局变量加锁同步实现通讯，不利于多个协程对全局变量的读写操作

channel特点

通道用于在goroutines之间共享数据，保证同步交换。channel需要指定数据类型，数据在channel上传递：任何时刻只有一个goroutine可以访问数据项，保证线程同步。channel底层是一个队列，线程安全的，多个协程并发访问时不需要加锁；channel是有类型的，一个string的channel只能存放string类型

channel声明和使用

var intChan chan int
chanMap := make(chan map[string]string, 10)
var chan1 chan Person
var chan2 chan *PersonintChan <- 10 //写入数据到channel
num := <- intChan // 读取channel的数据

channel是引用类型，必须**初始化(make)**才能使用。channel不能进行扩容，在没有使用协程的情况下，如果channel数据已取完，再取则直接报错 dead lock error

channel的接收特性

• 读和写操作对元素值的处理必须是原子性的
• 对于同一个channel同时写和读，即使写入速度快于读取速度，依旧不会造成阻塞dead lock

channel关闭和遍历

• 使用内置函数close关闭channel，当channel关闭后不能再写数据只能读取
• for-range遍历时，如果出现channel没有关闭则出现dead lock

只写channel和只读channel

channel可以声明为只读或者只写，默认是可读可写的

var intchan chan <- int // 只写channel
var intchan <-chan int  // 只读channel

3.2. 同步器

WaitGroup实现同步

由于主线程一旦执行完毕，无论goroutines是否执行完，整个程序都会结束。因此，需要一种同步机制来协调主线程和协程之间的执行顺序

WaitGroup类似于JUC中的CountDownLatch

• WaitGroup.Done() 表示已经完成了一个任务，等价于WaitGroup.Add(-1)
• WaitGroup.Add(1) 表示增加一个任务到协程队列，计数器+1
• 主线程中使用WaitGroup.Wait()，运行到这步会发生阻塞，直到WaitGroup中的计数器为0时才能继续向下执行

var wg sync.WaitGroupfunc main(){for i := 0; i < 10; i++ {go show(i)wg.Add(1)}wg.Wait() // 等价于countDownLatch.await();fmt.Println("[main] continue...")
}func task(i int) {// defer wg.Add(-1)defer wg.Done() // 等价于countDownLatch.countDown();fmt.Printf("[goroutine] 当前i=%d\n", i)
}

runtime包

• runtime.Gosched() 让出CPU时间片，重新等待安排任务

func printMsg(msg string) {for i := 0; i < 5; i++ {fmt.Printf("[goroutine] msg: %v\n", msg)}
}func main(){go printMsg("java is the best!")// go printMsg("spring cloud is all you need!")for i := 0; i < 5; i++ {runtime.Gosched()fmt.Println("[main] golang concurrent...")}fmt.Println("[main] continue...")
}

每次主线程运行到runtime.Gosched()时，将CPU时间片交出去，因此go printMsg任务会先执行，打印5次之后，主线程再打印5次。

• runtime.Goexit() 退出当前协程
• runtime.GOMAXPROCS 默认使用本机的最大CPU核数
• sync.Mutex 互斥锁

var (FactorialMap = make(map[int]uint64, 16)// 声明一个全局互斥锁lock sync.Mutex
)func main(){// 向map中写入数据for i := 1; i <= 20; i++ {go factorial(i)}// 防止主线程执行完毕goroutine还没完成任务time.Sleep(time.Second * 3)// 防止主线程和协程对临界资源的读写并发lock.Lock()for i, v := range FactorialMap {fmt.Printf("map[%d]=%d\n", i, v)}lock.Unlock()
}func factorial(n int) {var res uint64res = 1for i := 1; i <= n; i++ {res *= uint64(i)}// 存在并发写问题 -> concurrent map writes// 需要加入互斥锁lock.Lock()FactorialMap[n] = reslock.Unlock()
}

select和switch

• select用于处理异步IO操作，select可以监听case语句中channel的读写操作，当case中channel读写操作为非阻塞状态时（可读可写），触发相应的动作。解决从管道读取数据的阻塞问题，在遍历channel时如果不关闭则会发生阻塞导致deadlock

• select中的case语句必须是一个channel操作，default语句总是可执行的

• 如果有多个case都可运行，select可随机选出一个执行

• 如果没有case可以执行，那么执行default逻辑

• 如果没有case可以执行且没有default语句，select将会阻塞，直到某个case可以执行

var (intChan = make(chan int)strChan = make(chan string)
)
func main(){go func() {intChan <- 100strChan <- "golang"defer close(intChan)defer close(strChan)}()for {select {case r := <-intChan:fmt.Printf("[int chan] r: %v\n", r)case r := <-strChan:fmt.Printf("[string chan] r: %v\n", r)default:fmt.Println("no channel can be read!")}}fmt.Println("[main] continue...")
}

Timer

定时器，用于实现一些定时操作，内部也是通过channel实现的

func main(){timer1 := time.NewTimer(time.Second * 2)t1 := time.Now()fmt.Printf("time1: %v\n", t1)// timer1.C阻塞，直至2s结束继续执行t2 := <-timer1.Cfmt.Printf("time2: %v\n", t2)timer2 := time.NewTimer(time.Second * 2)<-timer2.Cfmt.Println("2s 后...")fmt.Printf("time3: %v\n", time.Now())timer3 := time.NewTimer(time.Second)go func() {<-timer3.Cfmt.Println("timer3 blocked!")}()// 定时器停止，上面匿名函数中的<-timer3.C就不会阻塞了stop := timer3.Stop()if stop {fmt.Println("timer3 stopped!")}
}

Ticker

Timer只会执行一次，Ticker可以周期性的执行

func main(){ticker := time.NewTicker(time.Second)var sum intintChan := make(chan int)// 每隔1s向intChan中写入一个数，select语句从三个case分支随机选择一个执行// 主线程一直在读取，直到sum>=10读取结束go func() {for _ = range ticker.C {select {case intChan <- 1:fmt.Println("int channel写入1")case intChan <- 2:fmt.Println("int channel写入2")case intChan <- 3:fmt.Println("int channel写入3")}}}()for v := range intChan {fmt.Println("从int channel中读取到: ", v)sum += vif sum >= 10 {break}}
}

原子变量

类似于JUC中的AtomicInteger原子整型等，使用CAS机制进行同步。常见原子操作有

• 加减 atomic.AddInt32(&num, 1)
• read atomic.LoadInt32(&num)
• CAS atomic.CompareAndSwapInt32(&num, 100, 200) 如果num==100修改为200否则此次CAS失败
• 交换 atomic.SwapInt32(&num, 200)
• write atomic.StoreInt32(&num, 200)

var num int32func AtomicTest() {for i := 0; i < 100; i++ {go add()go sub()}fmt.Printf("num: %v\n", num)
}func add() {atomic.AddInt32(&num, 1)fmt.Printf("[add method] num: %v\n", num)
}func sub() {atomic.AddInt32(&num, -1)fmt.Printf("[sub method] num: %v\n", num)
}