go-goroutine 和 channel
goroutine 和 channel
goroutine-看一个需求
需求:要求统计 1-9000000000 的数字中,哪些是素数?
分析思路:
1) 传统的方法,就是使用一个循环,循环的判断各个数是不是素数。[很慢]
2) 使用并发或者并行的方式,将统计素数的任务分配给多个 goroutine 去完成,这时就会使用到goroutine.【速度提高 4 倍】
goroutine-基本介绍
进程和线程介绍
程序、进程和线程的关系
并发和并行
并发和并行
1) 多线程程序在单核上运行,就是并发
2) 多线程程序在多核上运行,就是并行
go 协程和 go 主线程
go 主线程(有程序员直接称为线程/也可以理解成进程): 一个 go 线程上,可以起多个协程,你可以这样理解,协程是轻量级的线程[编译器做优化]。
go 协程的特点
1) 有独立的栈空间
2) 共享程序堆空间
3) 调度由用户控制
4) 协程是轻量级的线程
goroutine-快速入门
案例说明
请编写一个程序,完成如下功能:
1) 在主线程(可以理解成进程)中,开启一个 goroutine, 该协程每隔 1 秒输出 "hello,world"
2) 在主线程中也每隔一秒输出"hello,golang", 输出 10 次后,退出程序
3) 要求主线程和 goroutine 同时执行.
package main import ( "fmt" "strconv" "time" ) // 在主线程(可以理解成进程)中,开启一个goroutine, 该协程每隔1秒输出 "hello,world" // 在主线程中也每隔一秒输出"hello,golang", 输出10次后,退出程序 // 要求主线程和goroutine同时执行 //编写一个函数,每隔1秒输出 "hello,world" func test() { for i := 1; i <= 10; i++ { fmt.println("tesst () hello,world " + strconv.itoa(i)) time.sleep(time.second) } } func main() { go test() // 开启了一个协程 for i := 1; i <= 10; i++ { fmt.println(" main() hello,golang" + strconv.itoa(i)) time.sleep(time.second) } }
快速入门小结
1) 主线程是一个物理线程,直接作用在 cpu 上的。是重量级的,非常耗费 cpu 资源。
2) 协程从主线程开启的,是轻量级的线程,是逻辑态。对资源消耗相对小。
3) golang 的协程机制是重要的特点,可以轻松的 开启上万个协程。其它编程语言的并发机制是一般基于线程的,开启过多的线程,资源耗费大,这里就突显 golang 在并发上的优势了
goroutine 的调度模型
mpg 模式基本介绍
解释一下mpg含义:
m(machine):操作系统的主线程
p(processor):协程执行需要的资源(上下文context),可以看作一个局部的调度器,使go代码在一个线程上跑,他是实现从n:1到n:m映射的关键
g(gorountine):协程,有自己的栈。包含指令指针(instruction pointer)和其它信息(正在等待的channel等等),用于调度。一个p下面可以有多个g
mpg 模式运行的状态一
p的数量可以通过gomaxprocs()来设置,他其实代表了真正的并发度,即有多少个goroutine可以同时运行。p同时也维护着g(协程)的队列(称之为runqueue进程队列)。go代码中的m每有一个语句被执行,p就在末尾加入一个g(从runqueue队列中取出来的),在下一个调度点(p),就从runqueue队列中取出g。
p可以在os线程(主线程,或者是m)被阻塞时,转到另一个os线程(m)!go中的调度器保证有足够的线程来运行所有的p。当启用一个m0中的g0被syscall(系统调用)的时候,m0下面的p转给另一个线程m1(可以是创建的,也可以是原本就存在的)。m1接受了p(包括p所带的runqueue的队列里面所有状态的g,但不包括已经被syscall的g0),继续运行。而m0会等待执行syscall的g0的返回值。当g0的syscall结束后,他的主线程m0会尝试取得一个p来运行g0,一般情况下,他会从其他的m里面偷一个p过来,如果没有偷到的话就会把g0放到一个global runqueue(全局进程队列)中,然后把自己(m0)放进线程池或者转为休眠状态。
设置 golang 运行的 cpu 数
介绍:为了充分了利用多 cpu 的优势,在 golang 程序中,设置运行的 cpu 数目
package main import ( "runtime" "fmt" ) func main() { cpunum := runtime.numcpu() fmt.println("cpunum=", cpunum) //可以自己设置使用多个cpu runtime.gomaxprocs(cpunum - 1) fmt.println("ok") }
channel(管道)-看个需求
需求:现在要计算 1-200 的各个数的阶乘,并且把各个数的阶乘放入到 map 中。最后显示出来。
要求使用 goroutine 完成
分析思路:
1) 使用 goroutine 来完成,效率高,但是会出现并发/并行安全问题.
2) 这里就提出了不同 goroutine 如何通信的问题
代码实现
1) 使用 goroutine 来完成(看看使用 gorotine 并发完成会出现什么问题? 然后去解决)
2) 在运行某个程序时,如何知道是否存在资源竞争问题。 方法很简单,在编译该程序时,增加一个参数 -race 即可
不同 goroutine 之间如何通讯
1) 全局变量的互斥锁
2) 使用管道 channel 来解决
使用全局变量加锁同步改进程序
因为没有对全局变量 m 加锁,因此会出现资源争夺问题,代码会出现错误,提示 concurrent map
writes
解决方案:加入互斥锁
我们的数的阶乘很大,结果会越界,可以将求阶乘改成 sum += uint64(i)
package main import ( "fmt" _ "time" "sync" ) // 需求:现在要计算 1-200 的各个数的阶乘,并且把各个数的阶乘放入到map中。 // 最后显示出来。要求使用goroutine完成 // 思路 // 1. 编写一个函数,来计算各个数的阶乘,并放入到 map中. // 2. 我们启动的协程多个,统计的将结果放入到 map中 // 3. map 应该做出一个全局的. var ( mymap = make(map[int]int, 10) //声明一个全局的互斥锁 //lock 是一个全局的互斥锁, //sync 是包: synchornized 同步 //mutex : 是互斥 lock sync.mutex ) // test 函数就是计算 n!, 让将这个结果放入到 mymap func test(n int) { res := 1 for i := 1; i <= n; i++ { res *= i } //这里我们将 res 放入到mymap //加锁 lock.lock() mymap[n] = res //concurrent map writes? //解锁 lock.unlock() } func main() { // 我们这里开启多个协程完成这个任务[200个] for i := 1; i <= 20; i++ { go test(i) } //休眠10秒钟【第二个问题 】 //time.sleep(time.second * 5) //这里我们输出结果,变量这个结果 lock.lock() for i, v := range mymap { fmt.printf("map[%d]=%d\n", i, v) } lock.unlock() }
为什么需要 channel
1) 前面使用全局变量加锁同步来解决 goroutine 的通讯,但不完美
2) 主线程在等待所有 goroutine 全部完成的时间很难确定,我们这里设置 10 秒,仅仅是估算。
3) 如果主线程休眠时间长了,会加长等待时间,如果等待时间短了,可能还有 goroutine 处于工作
状态,这时也会随主线程的退出而销毁
4) 通过全局变量加锁同步来实现通讯,也并不利用多个协程对全局变量的读写操作。
5) 上面种种分析都在呼唤一个新的通讯机制-channel
channel 的基本介绍
1) channle 本质就是一个数据结构-队列
2) 数据是先进先出【fifo : first in first out】
3) 线程安全,多 goroutine 访问时,不需要加锁,就是说 channel 本身就是线程安全的
4) channel 有类型的,一个 string 的 channel 只能存放 string 类型数据。
定义/声明 channel
var 变量名 chan 数据类型
举例:
var intchan chan int (intchan 用于存放 int 数据) var mapchan chan map[int]string (mapchan 用于存放 map[int]string 类型) var perchan chan person var perchan2 chan *person ...
说明
channel 是引用类型
channel 必须初始化才能写入数据, 即 make 后才能使用
管道是有类型的,intchan 只能写入 整数 int
package main import ( "fmt" ) func main() { //演示一下管道的使用 //1. 创建一个可以存放3个int类型的管道 var intchan chan int intchan = make(chan int, 3) //2. 看看intchan是什么 fmt.printf("intchan 的值=%v intchan本身的地址=%p\n", intchan, &intchan) //3. 向管道写入数据 intchan<- 10 num := 211 intchan<- num intchan<- 50 // //如果从channel取出数据后,可以继续放入 <-intchan intchan<- 98//注意点, 当我们给管写入数据时,不能超过其容量 //4. 看看管道的长度和cap(容量) fmt.printf("channel len= %v cap=%v \n", len(intchan), cap(intchan)) // 3, 3 //5. 从管道中读取数据 var num2 int num2 = <-intchan fmt.println("num2=", num2) fmt.printf("channel len= %v cap=%v \n", len(intchan), cap(intchan)) // 2, 3 //6. 在没有使用协程的情况下,如果我们的管道数据已经全部取出,再取就会报告 deadlock num3 := <-intchan num4 := <-intchan //num5 := <-intchan fmt.println("num3=", num3, "num4=", num4/*, "num5=", num5*/) }
管道的初始化,写入数据到管道,从管道读取数据及基本的注意事项
channel 使用的注意事项
1) channel 中只能存放指定的数据类型
2) channle 的数据放满后,就不能再放入了
3) 如果从 channel 取出数据后,可以继续放入
4) 在没有使用协程的情况下,如果 channel 数据取完了,再取,就会报 dead lock
读写 channel 案例演示
package main import ( "fmt" ) type cat struct { name string age int } func main() { //定义一个存放任意数据类型的管道 3个数据 //var allchan chan interface{} allchan := make(chan interface{}, 3) allchan<- 10 allchan<- "tom jack" cat := cat{"小花猫", 4} allchan<- cat //我们希望获得到管道中的第三个元素,则先将前2个推出 <-allchan <-allchan newcat := <-allchan //从管道中取出的cat是什么? fmt.printf("newcat=%t , newcat=%v\n", newcat, newcat) //下面的写法是错误的!编译不通过 //fmt.printf("newcat.name=%v", newcat.name) //使用类型断言 a := newcat.(cat) fmt.printf("newcat.name=%v", a.name) }
channel 的遍历和关闭
channel 的关闭
使用内置函数 close 可以关闭 channel, 当 channel 关闭后,就不能再向 channel 写数据了,但是仍然可以从该 channel 读取数据
channel 的遍历
channel 支持 for--range 的方式进行遍历,请注意两个细节
1) 在遍历时,如果 channel 没有关闭,则回出现 deadlock 的错误
2) 在遍历时,如果 channel 已经关闭,则会正常遍历数据,遍历完后,就会退出遍历。
channel 遍历和关闭的案例演示
package main import ( "fmt" ) func main() { intchan := make(chan int, 3) intchan<- 100 intchan<- 200 close(intchan) // close //这是不能够再写入数到channel //intchan<- 300 fmt.println("okook~") //当管道关闭后,读取数据是可以的 n1 := <-intchan fmt.println("n1=", n1) //遍历管道 intchan2 := make(chan int, 100) for i := 0; i < 100; i++ { intchan2<- i * 2 //放入100个数据到管道 } //遍历管道不能使用普通的 for 循环 // for i := 0; i < len(intchan2); i++ { // } //在遍历时,如果channel没有关闭,则会出现deadlock的错误 //在遍历时,如果channel已经关闭,则会正常遍历数据,遍历完后,就会退出遍历 close(intchan2) for v := range intchan2 { fmt.println("v=", v) } }
应用示例--channel与goroutine
package main import ( "fmt" "time" ) //write data func writedata(intchan chan int) { for i := 1; i <= 50; i++ { //放入数据 intchan<- i // fmt.println("writedata ", i) //time.sleep(time.second) } close(intchan) //关闭 } //read data func readdata(intchan chan int, exitchan chan bool) { for { v, ok := <-intchan if !ok { break } time.sleep(time.second) fmt.printf("readdata 读到数据=%v\n", v) } //readdata 读取完数据后,即任务完成 exitchan<- true close(exitchan) } func main() { //创建两个管道 intchan := make(chan int, 10) exitchan := make(chan bool, 1) go writedata(intchan) go readdata(intchan, exitchan) time.sleep(time.second * 10) for { _, ok := <-exitchan if !ok { break } } }
应用实例 2-阻塞
若上面的代码,注释掉go readdata(intchan, exitchan)
,会怎样,因为管道有长度,所以当编译器发现一个管道只有写而没有读,改管道会阻塞(读与写的频率不一致没关系)!
应用实例 3
需求:
要求统计 1-200000 的数字中,哪些是素数?这个问题在本章开篇就提出了,现在我们有 goroutine和 channel 的知识后,就可以完成了 [测试数据: 80000]
分析思路:
传统的方法,就是使用一个循环,循环的判断各个数是不是素数【ok】。
使用并发/并行的方式,将统计素数的任务分配给多个(4 个)goroutine 去完成,完成任务时间短。
传统方法,一个协程
package main import ( "time" "fmt" ) func main() { start := time.now().unix() for num := 1; num <= 80000; num++ { flag := true //假设是素数 //判断num是不是素数 for i := 2; i < num; i++ { if num % i == 0 {//说明该num不是素数 flag = false break } } if flag { //将这个数就放入到primechan //primechan<- num } } end := time.now().unix() fmt.println("普通的方法耗时=", end - start) }
开了四个协程
package main import ( "fmt" "time" ) //向 intchan放入 1-8000个数 func putnum(intchan chan int) { for i := 1; i <= 80000; i++ { intchan<- i } //关闭intchan close(intchan) } // 从 intchan取出数据,并判断是否为素数,如果是,就 // //放入到primechan func primenum(intchan chan int, primechan chan int, exitchan chan bool) { //使用for 循环 // var num int var flag bool // for { //time.sleep(time.millisecond * 10) num, ok := <-intchan //intchan 取不到.. if !ok { break } flag = true //假设是素数 //判断num是不是素数 for i := 2; i < num; i++ { if num % i == 0 {//说明该num不是素数 flag = false break } } if flag { //将这个数就放入到primechan primechan<- num } } fmt.println("有一个primenum 协程因为取不到数据,退出") //这里我们还不能关闭 primechan //向 exitchan 写入true exitchan<- true } func main() { intchan := make(chan int , 1000) primechan := make(chan int, 20000)//放入结果 //标识退出的管道 exitchan := make(chan bool, 8) // 4个 start := time.now().unix() //开启一个协程,向 intchan放入 1-8000个数 go putnum(intchan) //开启4个协程,从 intchan取出数据,并判断是否为素数,如果是,就 //放入到primechan for i := 0; i < 8; i++ { go primenum(intchan, primechan, exitchan) } //这里我们主线程,进行处理 //直接 go func(){ for i := 0; i < 8; i++ { <-exitchan } end := time.now().unix() fmt.println("使用协程耗时=", end - start) //当我们从exitchan 取出了4个结果,就可以放心的关闭 prprimechan close(primechan) }() //遍历我们的 primechan ,把结果取出 for { _, ok := <-primechan if !ok{ break } //将结果输出 //fmt.printf("素数=%d\n", res) } fmt.println("main线程退出") }
结论:使用 go 协程后,执行的速度,理论上比普通方法提高至少 4 倍(我这是两倍)
channel 使用细节和注意事项
1) channel 可以声明为只读,或者只写性质 【案例演示】
package main import ( "fmt" ) func main() { //管道可以声明为只读或者只写 //1. 在默认情况下下,管道是双向 //var chan1 chan int //可读可写 //2 声明为只写 var chan2 chan<- int chan2 = make(chan int, 3) chan2<- 20 //num := <-chan2 //error fmt.println("chan2=", chan2) //3. 声明为只读 var chan3 <-chan int num2 := <-chan3 //chan3<- 30 //err fmt.println("num2", num2) }
3) 使用 select 可以解决从管道取数据的阻塞问题
package main import ( "fmt" "time" ) func main() { //使用select可以解决从管道取数据的阻塞问题 //1.定义一个管道 10个数据int intchan := make(chan int, 10) for i := 0; i < 10; i++ { intchan<- i } //2.定义一个管道 5个数据string stringchan := make(chan string, 5) for i := 0; i < 5; i++ { stringchan <- "hello" + fmt.sprintf("%d", i) } //传统的方法在遍历管道时,如果不关闭会阻塞而导致 deadlock //问题,在实际开发中,可能我们不好确定什么关闭该管道. //可以使用select 方式可以解决 //label: for { select { //注意: 这里,如果intchan一直没有关闭,不会一直阻塞而deadlock //,会自动到下一个case匹配 case v := <-intchan : fmt.printf("从intchan读取的数据%d\n", v) time.sleep(time.second) case v := <-stringchan : fmt.printf("从stringchan读取的数据%s\n", v) time.sleep(time.second) default : fmt.printf("都取不到了,不玩了, 程序员可以加入逻辑\n") time.sleep(time.second) return //break label } } }
4) goroutine 中使用 recover,解决协程中出现 panic,导致程序崩溃问题
如果我们开了一个协程,但这个协程出现panic,就会导致整个程序崩溃,这时我们可以在goroutine中使用recover来捕获panic,这样及时协程发生问题,主线程依然不受影响
package main import ( "fmt" "time" ) //函数 func sayhello() { for i := 0; i < 10; i++ { time.sleep(time.second) fmt.println("hello,world") } } //函数 func test() { //这里我们可以使用defer + recover defer func() { //捕获test抛出的panic if err := recover(); err != nil { fmt.println("test() 发生错误", err) } }() //定义了一个map var mymap map[int]string mymap[0] = "golang" //error } func main() { go sayhello() go test() for i := 0; i < 10; i++ { fmt.println("main() ok=", i) time.sleep(time.second) } }