关于Goroutine的原理

2024-04-29 16:18
文章标签 原理 goroutine

本文主要是介绍关于Goroutine的原理,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

原理上的内容比较多,比如goroutine启动的时候要执行哪些相关的操作,一点一点的补充一下。

channel的基本原理

channel是go语言中的特殊的机制,既可以同步两个被并发执行的函数,又可以让这两个函数通过相互传递特定类型的值来进行通信。事实上这也是channel的两个主要功能。

按照通道在初始化时是否有缓冲值,又可以分为缓冲通道非缓冲通道。通道初始化的时候也还是需要使用make进行,比如make(chan int,10)声明一个缓冲空间为10个int的通道,直接make(chan int)就是声明一个非缓冲通道

直接采用内建函数close(strChan)就可以关闭通道。应该保证在安全的情况下进行关闭通道的操作。基本的原则:内建函数 len(strChan)可以查看通道中当前有的元素的数量 cap(strChan)可以查看通道的总的容量,总容量一旦初始化之后就不会再发生改变了。

  • 无论怎样都不应该在接收端关闭通道,因为无法判断发送端是否还有数据要发送,通道有一个很好的特性,就是发送端关闭通道后,接收端仍然可以正常接受已经存在通道中的数据。谁启的通道,谁最后负责关,是这个道理。
  • 注意element , ok := <-chann 的这种语法, 如果通道被关闭则ok的值会变为false,element的值会变为该通道类型的零值,通常用ok这种语法来判断是否退出某个循环。

比如下面这段代码,同时也可以看下goroutine的相关使用模式:

package mainimport ("fmt""time"
)func main() {ch := make(chan int, 1)sign := make(chan byte, 2)go func() {for i := 0; i < 5; i++ {ch <- itime.Sleep(1 * time.Second)}close(ch)fmt.Println("The channel is closed.")sign <- 0}()go func() {//这个循环会一直尝试从ch中读取信息出来 即使ch已经被发送端关闭//但还是可以读信息出来 最后当ok 为false的时候 说明已经没有数据从ch中读出//跳出循环 注意这种判断方式for {fmt.Printf("before extract channel len: %v ,", len(ch))e, ok := <-chfmt.Printf("channel value: %d if extract ok :(%v) after extraction channel len : %v channel cap : %v \n", e, ok, len(ch), cap(ch))if !ok {break}time.Sleep(2 * time.Second)}fmt.Println("Done.")sign <- 1}()//要是不添加两次取值的操作的话 主进程就会马上结束 这里相当于是实现了一个//同步的操作 等待两个go func都结束之后 再结束主进程 注意这种技巧<-sign<-sign
}/*output:
before extract channel len: 1 ,channel value: 0 if extract ok :(true) after extraction channel len : 0 channel cap : 1
before extract channel len: 1 ,channel value: 1 if extract ok :(true) after extraction channel len : 0 channel cap : 1
before extract channel len: 1 ,channel value: 2 if extract ok :(true) after extraction channel len : 0 channel cap : 1
before extract channel len: 1 ,channel value: 3 if extract ok :(true) after extraction channel len : 0 channel cap : 1
The channel is closed.
before extract channel len: 1 ,channel value: 4 if extract ok :(true) after extraction channel len : 0 channel cap : 1
before extract channel len: 0 ,channel value: 0 if extract ok :(false) after extraction channel len : 0 channel cap : 1
Done.
*/
关于通道的基本原则
  • 通道缓冲已经满了的时候,再向通道中发送数据,会造成Goroutine的阻塞,通道没有初始化,即值为nil的时候,向其中发送数据会造成通道永久阻塞。
  • 关闭通道的操作应该由发送端来进行,通道关闭后,如果还有数据,接收端仍可以正常接受数据。
  • 向通道中发送值,进行的是值传递

channel使用场景分析

使用场景(1)

注意 app.go文件夹中的 346 行左右开始地方的一个坑 注意time.After的返回值 由于放在了for循环中 因此 每次会新new 一个 channel出来 还有注意跳出多层循环的方式
主要参考的是《Go并发编程实战的相关内容》

代码如下:

package mainimport ("fmt""runtime"
)func main() {names := []string{"E", "H", "R", "J", "M"}for _, name := range names {go func() {fmt.Printf("Hello , %s \n", name)}()}//要是不添加runtime的话 就不会有内容输出//因为for循环执行速度太快了 直接循环结束跳出了最后的循环//之后 for循环中生成的5个go func 会被分别进行调度runtime.Gosched()
}/* output
Hello , M 
Hello , M 
Hello , M 
Hello , M 
Hello , M
*/

根据代码可以看出,具体循环的时候for循环中的go func 的调度并不是按照想象的那样,一次循环一个go func ,不要对go func的执行时机做任何假设。

优化方案

一种思路是把runtime.Gosched()函数放在每次for循环结束的时候,这样每次for循环之后,都会被重新调度一次,可能会出现正确的结果,并不是每次都准确,比如go func的程序需要运行一段时间,在这段运行的时间之内,可能就已经循环了几个元素过去了

package mainimport ("fmt""runtime""time"
)func main() {names := []string{"E", "H", "R", "J", "M", "N", "O", "P"}for _, name := range names {go func() {time.Sleep(1000 * time.Nanosecond)fmt.Printf("Hello , %s \n", name)}()runtime.Gosched()}}/* output:
Hello , E
Hello , J
Hello , J
Hello , P
Hello , P
Hello , P
*/

还有一种思路是采用传递参数的方式,就是给goroutine带上了参数,虽然goroutine已经脱离了main函数的控制,但是它已经带上了main函数给的烙印,相当于是某种解耦的感觉,for循环多次就不会影响打印的结果了,比如下面代码:

package mainimport ("fmt""runtime""time"
)func main() {names := []string{"E", "H", "R", "J", "M", "N", "O", "P"}for _, name := range names {go func(who string) {time.Sleep(1000 * time.Nanosecond)fmt.Printf("Hello , %s \n", who)}(name)}runtime.Gosched()}/* output:
Hello , E
Hello , H
Hello , R
Hello , J
Hello , M
*/

但是这个方法仍然很有问题,只能保证在函数执行时间很短的时候结果正常,而且不输出重复的内容,如果程序执行时间比较长的话,很有可能main函数会被提前结束,按顺序生成的多个goroutine在cpu那里会不会仍然按照顺序被调度执行?这个仍然不确定?有几个goroutine会不能被正常调度到并且执行,比如像上面的代码的输出样子,而且每次输出的结果也都是不确定的。

使用场景(2)

编码的时候遇到这样一个场景,服务创建成功之后,需要等待ip被分配,ip被分配好之后,服务才正式部署成功,最后将所有的信息返回给前台,于是打算这样实现,在服务创建成功之后就不断循环,查询ip如果分配成功了就返回,如果超过了时间也返回失败,最后这部分的代码像下面这样:

//一个时间控制的channel
//注意这个要在循环之外单独声明 否则每次都会分配一个新的 time.After的channel返回过来
t := time.After(time.Second * 10)//注意这种跳出多层循环的操作方式 要是单层使用break的话 仅仅跳出的是 select 那一层的循环A:for {select {//如果时间到了 就返回错误信息case <-t:log.Println("time out to allocate ip")//delete the se which deploy faileda.Ctx.ResponseWriter.Header().Set("Content-Type", "application/json")http.Error(a.Ctx.ResponseWriter, `{"errorMessage":"`+"deploy error : time out"+`"}`, 406)break A//如果时间没到 就是 t 还没有发回信息 select语句就默认跳转到default块中//执行查找ip是否分配的操作default://log.Println("logout:", <-timeout)sename := service.ObjectMeta.Labels["name"]podslist, err := a.Podip(sename)if err != nil {log.Println(err.Error())a.Ctx.ResponseWriter.Header().Set("Content-Type", "application/json")http.Error(a.Ctx.ResponseWriter, `{"errorMessage":"`+err.Error()+`"}`, 406)break A}if len(podslist) == 0 {continue} else {log.Println("allocation ok ......")a.Data["json"] = detaila.ServeJson()break A}}}
使用场景(3)

常常有这样一种场景,把某些信息从旧的资源池中取出来,经过一些加工处理,再放入新的资源池中,这个过程如果按传统的方式就是采用完全串行的方式效率会很低,粒度太粗了,具体的粒度可以细化以每次所取的单位资源为粒度。
比如以书上p339为例,有一个资源池存储这person的信息,将每个person从中取出来,之后进行一些处理,再存到新的资源池中,这里用oldarray以及newarray来模拟旧的和新的资源池:

具体的代码如下:

package main//参考go 并发编程实战 p337
import ("log""strconv""time"
)type Person struct {name stringage  intaddr string
}var oldpersonarray = [5]Person{}
var newpersonarray = [5]Person{}type PersonHandler interface {Batch(origs <-chan Person) <-chan PersonHandle(orig *Person)
}//struct 实现了personhandler 接口
type PersonHandlerImpl struct{}//从origs接收信息 处理之后再返回给新的channel
func (handler PersonHandlerImpl) Batch(origs <-chan Person) <-chan Person {dests := make(chan Person, 100)go func() {for {p, ok := <-origsif !ok {close(dests)break}handler.Handle(&p)log.Printf("old value : %v\n", p)//time.Sleep(time.Second)dests <- p}}()return dests
}//这里要使用引用传递
func (handler PersonHandlerImpl) Handle(orig *Person) {orig.addr = "new address"
}func getPersonHandler() PersonHandler {return &PersonHandlerImpl{}}//print the oldpersonarray into the chan<-Person
func fetchPerson(origs chan<- Person) {for _, v := range oldpersonarray {//fmt.Printf("get the value : %v %v \n", k, v)time.Sleep(time.Second)origs <- v}close(origs)}//fetch the value from the channel and store it into the newpersonarray
func savePerson(dest <-chan Person) <-chan int {intChann := make(chan int)go func() {index := 0for {p, ok := <-destif !ok {break}//time.Sleep(time.Second)log.Printf("new value transfer %v \n", p)newpersonarray[index] = pindex++}intChann <- 1}()return intChann
}func init() {//使用range的话是值传递 这里要给oldpersonarray赋值进来tmplen := len(oldpersonarray)for i := 0; i < tmplen; i++ {oldpersonarray[i].addr = "old address"oldpersonarray[i].age = ioldpersonarray[i].name = strconv.Itoa(i)}log.Printf("first print init value : %v\n", oldpersonarray)}
func main() {handeler := getPersonHandler()origs := make(chan Person, 100)dests := handeler.Batch(origs)//go func() { fetchPerson(origs) }()// 不加go func的话 要等这句执行完 才能执行下一句// 则orgis信息都输出 完全关闭掉 这个时候 从dest接收信息的语句才开始执行// 所以不会动态输出 这句加上go func的话 就会没隔 1s 动态输出// 如果将fetchPerson 再往前面放一句 则old value也不会动态输出fetchPerson(origs)sign := savePerson(dests)<-signlog.Printf("last print new value : %v \n", newpersonarray)}

整体的结构图如下:

代码结构

代码基本分析:

  • 首先声明一个 PersonHandler 的接口,之后声明一个struct PersonHandlerImpl 将接口中的两个方法都实现了,init函数用于进行oldarray的初始化工作。注意为了减少出错,内部的函数在方声明的时候都是单向的channel。
  • 1,2 fetchperson从oldarray中区数据,并把数据存到origs channel中,注意最后取完数据到通道之后,要由发送方将channel关闭,否则可能造成deadlock。注意在main函数中,如果fech操作没有放到一个goroutine中来执行,就仍然是串行的,相当于是把数据都放入到channel中,另一端才开始取,没发挥出并发的优势。
  • 3,4 Batch函数将person信息从origs中取出来,进行处理后,同时传到dests中,最后将dests返回,注意这里不是全部传入之后才将dests返回,而是新启动一个goroutine执行传入操作,同时将dests返回,注意要主动关闭channel。
  • 5 savePerson操作接收一个<-chann 之后从中接受person信息,将值写入到新的资源池中,最后全部写入结束之后,传一个sign channel给主进程,结束。
  • 总结,在需要动态输出信息的时候,goroutine往往是和channel结合在一起使用。最常见的用法是,一个goroutine负责向channel中写入数据,之后将channel返回,由其他进程取出信息。比如之前写过的一些websocket从前台接受信息,后台处理信息之后再动态返回给前台打出结果的模型,就和这个差不多,总之具体的异步执行流程要理清楚,都有哪些channel,负责传递的信息分别是什么。

这篇关于关于Goroutine的原理的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/946559

相关文章

Redis主从复制实现原理分析

《Redis主从复制实现原理分析》Redis主从复制通过Sync和CommandPropagate阶段实现数据同步,2.8版本后引入Psync指令,根据复制偏移量进行全量或部分同步,优化了数据传输效率... 目录Redis主DodMIK从复制实现原理实现原理Psync: 2.8版本后总结Redis主从复制实

深入探索协同过滤:从原理到推荐模块案例

文章目录 前言一、协同过滤1. 基于用户的协同过滤(UserCF)2. 基于物品的协同过滤(ItemCF)3. 相似度计算方法 二、相似度计算方法1. 欧氏距离2. 皮尔逊相关系数3. 杰卡德相似系数4. 余弦相似度 三、推荐模块案例1.基于文章的协同过滤推荐功能2.基于用户的协同过滤推荐功能 前言     在信息过载的时代,推荐系统成为连接用户与内容的桥梁。本文聚焦于

hdu4407(容斥原理)

题意:给一串数字1,2,......n,两个操作:1、修改第k个数字,2、查询区间[l,r]中与n互质的数之和。 解题思路:咱一看,像线段树,但是如果用线段树做,那么每个区间一定要记录所有的素因子,这样会超内存。然后我就做不来了。后来看了题解,原来是用容斥原理来做的。还记得这道题目吗?求区间[1,r]中与p互质的数的个数,如果不会的话就先去做那题吧。现在这题是求区间[l,r]中与n互质的数的和

hdu4407容斥原理

题意: 有一个元素为 1~n 的数列{An},有2种操作(1000次): 1、求某段区间 [a,b] 中与 p 互质的数的和。 2、将数列中某个位置元素的值改变。 import java.io.BufferedInputStream;import java.io.BufferedReader;import java.io.IOException;import java.io.Inpu

hdu4059容斥原理

求1-n中与n互质的数的4次方之和 import java.io.BufferedInputStream;import java.io.BufferedReader;import java.io.IOException;import java.io.InputStream;import java.io.InputStreamReader;import java.io.PrintWrit

寻迹模块TCRT5000的应用原理和功能实现(基于STM32)

目录 概述 1 认识TCRT5000 1.1 模块介绍 1.2 电气特性 2 系统应用 2.1 系统架构 2.2 STM32Cube创建工程 3 功能实现 3.1 代码实现 3.2 源代码文件 4 功能测试 4.1 检测黑线状态 4.2 未检测黑线状态 概述 本文主要介绍TCRT5000模块的使用原理,包括该模块的硬件实现方式,电路实现原理,还使用STM32类

TL-Tomcat中长连接的底层源码原理实现

长连接:浏览器告诉tomcat不要将请求关掉。  如果不是长连接,tomcat响应后会告诉浏览器把这个连接关掉。    tomcat中有一个缓冲区  如果发送大批量数据后 又不处理  那么会堆积缓冲区 后面的请求会越来越慢。

PHP原理之内存管理中难懂的几个点

PHP的内存管理, 分为俩大部分, 第一部分是PHP自身的内存管理, 这部分主要的内容就是引用计数, 写时复制, 等等面向应用的层面的管理. 而第二部分就是今天我要介绍的, zend_alloc中描写的关于PHP自身的内存管理, 包括它是如何管理可用内存, 如何分配内存等. 另外, 为什么要写这个呢, 因为之前并没有任何资料来介绍PHP内存管理中使用的策略, 数据结构, 或者算法. 而在我们

Smarty模板执行原理

为了实现程序的业务逻辑和内容表现页面的分离从而提高开发速度,php 引入了模板引擎的概念,php 模板引擎里面最流行的可以说是smarty了,smarty因其功能强大而且速度快而被广大php web开发者所认可。本文将记录一下smarty模板引擎的工作执行原理,算是加深一下理解。 其实所有的模板引擎的工作原理是差不多的,无非就是在php程序里面用正则匹配将模板里面的标签替换为php代码从而将两者

Restful API 原理以及实现

先说说API 再说啥是RESRFUL API之前,咱先说说啥是API吧。API大家应该都知道吧,简称接口嘛。随着现在移动互联网的火爆,手机软件,也就是APP几乎快爆棚了。几乎任何一个网站或者应用都会出一款iOS或者Android APP,相比网页版的体验,APP确实各方面性能要好很多。 那么现在问题来了。比如QQ空间网站,如果我想获取一个用户发的说说列表。 QQ空间网站里面需要这个功能。