【golang详解】go语言GMP(GPM)原理和调度

Python014

【golang详解】go语言GMP(GPM)原理和调度,第1张

Goroutine调度是一个很复杂的机制,下面尝试用简单的语言描述一下Goroutine调度机制,想要对其有更深入的了解可以去研读一下源码。

首先介绍一下GMP什么意思:

G ----------- goroutine: 即Go协程,每个go关键字都会创建一个协程。

M ---------- thread内核级线程,所有的G都要放在M上才能运行。

P ----------- processor处理器,调度G到M上,其维护了一个队列,存储了所有需要它来调度的G。

Goroutine 调度器P和 OS 调度器是通过 M 结合起来的,每个 M 都代表了 1 个内核线程,OS 调度器负责把内核线程分配到 CPU 的核上执行

模型图:

避免频繁的创建、销毁线程,而是对线程的复用。

1)work stealing机制

  当本线程无可运行的G时,尝试从其他线程绑定的P偷取G,而不是销毁线程。

2)hand off机制

  当本线程M0因为G0进行系统调用阻塞时,线程释放绑定的P,把P转移给其他空闲的线程执行。进而某个空闲的M1获取P,继续执行P队列中剩下的G。而M0由于陷入系统调用而进被阻塞,M1接替M0的工作,只要P不空闲,就可以保证充分利用CPU。M1的来源有可能是M的缓存池,也可能是新建的。当G0系统调用结束后,根据M0是否能获取到P,将会将G0做不同的处理:

如果有空闲的P,则获取一个P,继续执行G0。

如果没有空闲的P,则将G0放入全局队列,等待被其他的P调度。然后M0将进入缓存池睡眠。

如下图

GOMAXPROCS设置P的数量,最多有GOMAXPROCS个线程分布在多个CPU上同时运行

在Go中一个goroutine最多占用CPU 10ms,防止其他goroutine被饿死。

具体可以去看另一篇文章

【Golang详解】go语言调度机制 抢占式调度

当创建一个新的G之后优先加入本地队列,如果本地队列满了,会将本地队列的G移动到全局队列里面,当M执行work stealing从其他P偷不到G时,它可以从全局G队列获取G。

协程经历过程

我们创建一个协程 go func()经历过程如下图:

说明:

这里有两个存储G的队列,一个是局部调度器P的本地队列、一个是全局G队列。新创建的G会先保存在P的本地队列中,如果P的本地队列已经满了就会保存在全局的队列中;处理器本地队列是一个使用数组构成的环形链表,它最多可以存储 256 个待执行任务。

G只能运行在M中,一个M必须持有一个P,M与P是1:1的关系。M会从P的本地队列弹出一个可执行状态的G来执行,如果P的本地队列为空,就会想其他的MP组合偷取一个可执行的G来执行;

一个M调度G执行的过程是一个循环机制;会一直从本地队列或全局队列中获取G

上面说到P的个数默认等于CPU核数,每个M必须持有一个P才可以执行G,一般情况下M的个数会略大于P的个数,这多出来的M将会在G产生系统调用时发挥作用。类似线程池,Go也提供一个M的池子,需要时从池子中获取,用完放回池子,不够用时就再创建一个。

work-stealing调度算法:当M执行完了当前P的本地队列队列里的所有G后,P也不会就这么在那躺尸啥都不干,它会先尝试从全局队列队列寻找G来执行,如果全局队列为空,它会随机挑选另外一个P,从它的队列里中拿走一半的G到自己的队列中执行。

如果一切正常,调度器会以上述的那种方式顺畅地运行,但这个世界没这么美好,总有意外发生,以下分析goroutine在两种例外情况下的行为。

Go runtime会在下面的goroutine被阻塞的情况下运行另外一个goroutine:

用户态阻塞/唤醒

当goroutine因为channel操作或者network I/O而阻塞时(实际上golang已经用netpoller实现了goroutine网络I/O阻塞不会导致M被阻塞,仅阻塞G,这里仅仅是举个栗子),对应的G会被放置到某个wait队列(如channel的waitq),该G的状态由_Gruning变为_Gwaitting,而M会跳过该G尝试获取并执行下一个G,如果此时没有可运行的G供M运行,那么M将解绑P,并进入sleep状态;当阻塞的G被另一端的G2唤醒时(比如channel的可读/写通知),G被标记为,尝试加入G2所在P的runnext(runnext是线程下一个需要执行的 Goroutine。), 然后再是P的本地队列和全局队列。

系统调用阻塞

当M执行某一个G时候如果发生了阻塞操作,M会阻塞,如果当前有一些G在执行,调度器会把这个线程M从P中摘除,然后再创建一个新的操作系统的线程(如果有空闲的线程可用就复用空闲线程)来服务于这个P。当M系统调用结束时候,这个G会尝试获取一个空闲的P执行,并放入到这个P的本地队列。如果获取不到P,那么这个线程M变成休眠状态, 加入到空闲线程中,然后这个G会被放入全局队列中。

队列轮转

可见每个P维护着一个包含G的队列,不考虑G进入系统调用或IO操作的情况下,P周期性的将G调度到M中执行,执行一小段时间,将上下文保存下来,然后将G放到队列尾部,然后从队列中重新取出一个G进行调度。

除了每个P维护的G队列以外,还有一个全局的队列,每个P会周期性地查看全局队列中是否有G待运行并将其调度到M中执行,全局队列中G的来源,主要有从系统调用中恢复的G。之所以P会周期性地查看全局队列,也是为了防止全局队列中的G被饿死。

除了每个P维护的G队列以外,还有一个全局的队列,每个P会周期性地查看全局队列中是否有G待运行并将其调度到M中执行,全局队列中G的来源,主要有从系统调用中恢复的G。之所以P会周期性地查看全局队列,也是为了防止全局队列中的G被饿死。

M0

M0是启动程序后的编号为0的主线程,这个M对应的实例会在全局变量rutime.m0中,不需要在heap上分配,M0负责执行初始化操作和启动第一个G,在之后M0就和其他的M一样了

G0

G0是每次启动一个M都会第一个创建的goroutine,G0仅用于负责调度G,G0不指向任何可执行的函数,每个M都会有一个自己的G0,在调度或系统调用时会使用G0的栈空间,全局变量的G0是M0的G0

一个G由于调度被中断,此后如何恢复?

中断的时候将寄存器里的栈信息,保存到自己的G对象里面。当再次轮到自己执行时,将自己保存的栈信息复制到寄存器里面,这样就接着上次之后运行了。

我这里只是根据自己的理解进行了简单的介绍,想要详细了解有关GMP的底层原理可以去看Go调度器 G-P-M 模型的设计者的文档或直接看源码

参考: (https://www.cnblogs.com/X-knight/p/11365929.html)

(https://draveness.me/golang/docs/part3-runtime/ch06-concurrency/golang-goroutine/)

个人觉得golang十分适合进行网游服务器端开发,写下这篇文章总结一下。从网游的角度看:要成功的运营一款网游,很大程度上依赖于玩家自发形成的社区。只有玩家自发形成一个稳定的生态系统,游戏才能持续下去,避免鬼城的出现。而这就需要多次大量导入用户,在同时在线用户量达到某个临界点的时候,才有可能完成。因此,多人同时在线十分有必要。再来看网游的常见玩法,除了排行榜这类统计和数据汇总的功能外,基本没有需要大量CPU时间的应用。以前的项目里,即时战斗产生的各种伤害计算对CPU的消耗也不大。玩家要完成一次操作,需要通过客户端-服务器端-客户端这样一个来回,为了获得高响应速度,满足玩家体验,服务器端的处理也不能占用太多时间。所以,每次请求对应的CPU占用是比较小的。网游的IO主要分两个方面,一个是网络IO,一个是磁盘IO。网络IO方面,可以分成美术资源的IO和游戏逻辑指令的IO,这里主要分析游戏逻辑的IO。游戏逻辑的IO跟CPU占用的情况相似,每次请求的字节数很小,但由于多人同时在线,因此并发数相当高。另外,地图信息的广播也会带来比较频繁的网络通信。磁盘IO方面,主要是游戏数据的保存。采用不同的数据库,会有比较大的区别。以前的项目里,就经历了从MySQL转向MongoDB这种内存数据库的过程,磁盘IO不再是瓶颈。总体来说,还是用内存做一级缓冲,避免大量小数据块读写的方案。针对网游的这些特点,golang的语言特性十分适合开发游戏服务器端。首先,go语言提供goroutine机制作为原生的并发机制。每个goroutine所需的内存很少,实际应用中可以启动大量的goroutine对并发连接进行响应。goroutine与gevent中的greenlet很相像,遇到IO阻塞的时候,调度器就会自动切换到另一个goroutine执行,保证CPU不会因为IO而发生等待。而goroutine与gevent相比,没有了python底层的GIL限制,就不需要利用多进程来榨取多核机器的性能了。通过设置最大线程数,可以控制go所启动的线程,每个线程执行一个goroutine,让CPU满负载运行。同时,go语言为goroutine提供了独到的通信机制channel。channel发生读写的时候,也会挂起当前操作channel的goroutine,是一种同步阻塞通信。这样既达到了通信的目的,又实现同步,用CSP模型的观点看,并发模型就是通过一组进程和进程间的事件触发解决任务的。虽然说,主流的编程语言之间,只要是图灵完备的,他们就都能实现相同的功能。但go语言提供的这种协程间通信机制,十分优雅地揭示了协程通信的本质,避免了以往锁的显式使用带给程序员的心理负担,确是一大优势。进行网游开发的程序员,可以将游戏逻辑按照单线程阻塞式的写,不需要额外考虑线程调度的问题,以及线程间数据依赖的问题。因为,线程间的channel通信,已经表达了线程间的数据依赖关系了,而go的调度器会给予妥善的处理。另外,go语言提供的gc机制,以及对指针的保护式使用,可以大大减轻程序员的开发压力,提高开发效率。展望未来,我期待go语言社区能够提供更多的goroutine间的隔离机制。个人十分推崇erlang社区的脆崩哲学,推动应用发生预期外行为时,尽早崩溃,再fork出新进程处理新的请求。对于协程机制,需要由程序员保证执行的函数不会发生死循环,导致线程卡死。

一、Kafka简述

1. 为什么需要用到消息队列

异步:对比以前的串行同步方式来说,可以在同一时间做更多的事情,提高效率;

解耦:在耦合太高的场景,多个任务要对同一个数据进行操作消费的时候,会导致一个任务的处理因为另一个任务对数据的操作变得及其复杂。

缓冲:当遇到突发大流量的时候,消息队列可以先把所有消息有序保存起来,避免直接作用于系统主体,系统主题始终以一个平稳的速率去消费这些消息。

2.为什么选择kafka呢?

这没有绝对的好坏,看个人需求来选择,我这里就抄了一段他人总结的的优缺点,可见原文

kafka的优点:

1.支持多个生产者和消费者2.支持broker的横向拓展3.副本集机制,实现数据冗余,保证数据不丢失4.通过topic将数据进行分类5.通过分批发送压缩数据的方式,减少数据传输开销,提高吞高量6.支持多种模式的消息7.基于磁盘实现数据的持久化8.高性能的处理信息,在大数据的情况下,可以保证亚秒级的消息延迟9.一个消费者可以支持多种topic的消息10.对CPU和内存的消耗比较小11.对网络开销也比较小12.支持跨数据中心的数据复制13.支持镜像集群

kafka的缺点:

1.由于是批量发送,所以数据达不到真正的实时2.对于mqtt协议不支持3.不支持物联网传感数据直接接入4.只能支持统一分区内消息有序,无法实现全局消息有序5.监控不完善,需要安装插件6.需要配合zookeeper进行元数据管理7.会丢失数据,并且不支持事务8.可能会重复消费数据,消息会乱序,可用保证一个固定的partition内部的消息是有序的,但是一个topic有多个partition的话,就不能保证有序了,需要zookeeper的支持,topic一般需要人工创建,部署和维护一般都比mq高

3. Golang 操作kafka

3.1. kafka的环境

网上有很多搭建kafka环境教程,这里就不再搭建,就展示一下kafka的环境,在kubernetes上进行的搭建,有需要的私我,可以发yaml文件

3.2. 第三方库

github.com/Shopify/sarama // kafka主要的库*github.com/bsm/sarama-cluster // kafka消费组

3.3. 消费者

单个消费者

funcconsumer(){varwg sync.WaitGroup  consumer, err := sarama.NewConsumer([]string{"172.20.3.13:30901"},nil)iferr !=nil{      fmt.Println("Failed to start consumer: %s", err)return}  partitionList, err := consumer.Partitions("test0")//获得该topic所有的分区iferr !=nil{      fmt.Println("Failed to get the list of partition:, ", err)return}forpartition :=rangepartitionList {      pc, err := consumer.ConsumePartition("test0",int32(partition), sarama.OffsetNewest)iferr !=nil{        fmt.Println("Failed to start consumer for partition %d: %s\n", partition, err)return}      wg.Add(1)gofunc(sarama.PartitionConsumer){//为每个分区开一个go协程去取值formsg :=rangepc.Messages() {//阻塞直到有值发送过来,然后再继续等待fmt.Printf("Partition:%d, Offset:%d, key:%s, value:%s\n", msg.Partition, msg.Offset,string(msg.Key),string(msg.Value))        }deferpc.AsyncClose()        wg.Done()      }(pc)  }  wg.Wait()}funcmain(){  consumer()}

消费组

funcconsumerCluster(){  groupID :="group-1"config := cluster.NewConfig()  config.Group.Return.Notifications =trueconfig.Consumer.Offsets.CommitInterval =1* time.Second  config.Consumer.Offsets.Initial = sarama.OffsetNewest//初始从最新的offset开始c, err := cluster.NewConsumer(strings.Split("172.20.3.13:30901",","),groupID, strings.Split("test0",","), config)iferr !=nil{      glog.Errorf("Failed open consumer: %v", err)return}deferc.Close()gofunc(c *cluster.Consumer){      errors := c.Errors()      noti := c.Notifications()for{select{caseerr := <-errors:            glog.Errorln(err)case<-noti:        }      }  }(c)formsg :=rangec.Messages() {      fmt.Printf("Partition:%d, Offset:%d, key:%s, value:%s\n", msg.Partition, msg.Offset,string(msg.Key),string(msg.Value))      c.MarkOffset(msg,"")//MarkOffset 并不是实时写入kafka,有可能在程序crash时丢掉未提交的offset}}funcmain(){goconsumerCluster()}

3.4. 生产者

同步生产者

packagemainimport("fmt""github.com/Shopify/sarama")funcmain(){  config := sarama.NewConfig()  config.Producer.RequiredAcks = sarama.WaitForAll//赋值为-1:这意味着producer在follower副本确认接收到数据后才算一次发送完成。config.Producer.Partitioner = sarama.NewRandomPartitioner//写到随机分区中,默认设置8个分区config.Producer.Return.Successes =truemsg := &sarama.ProducerMessage{}  msg.Topic =`test0`msg.Value = sarama.StringEncoder("Hello World!")  client, err := sarama.NewSyncProducer([]string{"172.20.3.13:30901"}, config)iferr !=nil{      fmt.Println("producer close err, ", err)return}deferclient.Close()  pid, offset, err := client.SendMessage(msg)iferr !=nil{      fmt.Println("send message failed, ", err)return}  fmt.Printf("分区ID:%v, offset:%v \n", pid, offset)}

异步生产者

funcasyncProducer(){  config := sarama.NewConfig()  config.Producer.Return.Successes =true//必须有这个选项config.Producer.Timeout =5* time.Second  p, err := sarama.NewAsyncProducer(strings.Split("172.20.3.13:30901",","), config)deferp.Close()iferr !=nil{return}//这个部分一定要写,不然通道会被堵塞gofunc(p sarama.AsyncProducer){      errors := p.Errors()      success := p.Successes()for{select{caseerr := <-errors:iferr !=nil{              glog.Errorln(err)            }case<-success:        }      }  }(p)for{      v :="async: "+ strconv.Itoa(rand.New(rand.NewSource(time.Now().UnixNano())).Intn(10000))      fmt.Fprintln(os.Stdout, v)      msg := &sarama.ProducerMessage{        Topic: topics,        Value: sarama.ByteEncoder(v),      }      p.Input() <- msg      time.Sleep(time.Second *1)  }}funcmain(){goasyncProducer()select{      }}

3.5. 结果展示->

同步生产打印:

分区ID:0,offset:90

消费打印:

Partition:0,Offset:90,key:,value:Hello World!

异步生产打印:

async:7272async:7616async:998

消费打印:

Partition:0,Offset:91,key:,value:async:7272Partition:0,Offset:92,key:,value:async:7616Partition:0,Offset:93,key:,value:async:998