Go语言入门到实战——14.Go语言的协程机制以及并发机制

Go语言入门到实战——00主目录
在上一讲中我们学习了Go语言的包的知识已经依赖管理。
协程(groutine)是一种更加轻量级的线程(thread)。

一.协程与线程得到比较

1.对于java而言,在JDK5以后,线程创建时线程栈的默认大小为1M,而协程栈初始化栈大小为2K
2.两者与系统线程(KSE,Kernel Space Entity,内核对象)的对应关系,java的线程与KSE的
对应关系是1:1的,而java的协程与KSE的对应关系是M:N的

我们可以看到当采用KSE与线程是1：1的关系的时候，在调度的时候KSE由CPU直接调度，对应的去调度线程的话效率虽然很高，但是如果是需要进行切换的话，这个时候1：1就必须一个个的切换，切换次数多了对资源的消耗是非常大的，而如果选择多对多的话则不会有这么高的消耗。

上面是协程与系统线程的多对多关系图。我们可以看到系统线程调度协程时，会有一个协程队列，队列里面的协程都在处于等待状态,还有一个协程(即上面的G0)则是正在由系统线程调度运行。当当前运行的协程(G0)发生中断后(系统中断M0导致G0也中断),Processor就会选择另外一个系统线程,就是上面左图到右图的过程,继续执行队列里面的其他的协程,当协程被唤醒后就会重新加入到队列的尾部中(中断发生后寄存器的协程状态会被保存在协程对象,当再次执行时就会再次放入寄存器中)。那么这里为了防止出现饿死的现象,GO是这么设计的：

GO语言里面有一个守护线程在运行,他会记录Processor已经完成的协程的数量,当这个数量在
一段时间内都没有发生变化的时候,他就会设置一个标记,而目前正在运行的协程在读到这个标
记后就会把自己诊断下来,然后将自己插入到等待队列的队尾去,前面的协程进入执行。

那么go启动一个协程的方式是比较简单的，在执行的函数前面加上go关键字就可以启动一个协程了。

package testimport "testing"func TestGroutine(t *testing.T) {for i := 0; i < 10; i++ {go func(i int) {t.Log(i)}(i)}
}

package test
//这种情况协程共享i变量而造成打印很多一样的结果
import "testing"func TestGroutine(t *testing.T) {for i := 0; i < 10; i++ {go func() {t.Log(i)}()}
}

二.共享内存并发机制

package test
//很明显这段代码是线程不安全的,输出的counter的值是小于5000的
import ("testing""time"
)func TestGroutineThreadSafe(t *testing.T) {var counter int = 0for i := 0; i < 5000; i++ {go func() {counter++}()}time.Sleep(2 * time.Second)t.Log(counter)
}

上面出现这个情况的原因是由于每一个协程都是在共享Couter变量,但是没有加锁导致的。
下面我们使用锁机制来避免这一情况:

package test
//加锁之后counter的结果就正常了
import ("sync""testing""time"
)func TestGroutineThreadSafe(t *testing.T) {var mut sync.Mutexvar counter int = 0for i := 0; i < 5000; i++ {go func() {defer func() {mut.Unlock()}()mut.Lock()counter++}()}time.Sleep(1 * time.Second)t.Log(counter)
}

进行改进

package testimport ("sync""testing"
)func TestGroutineThreadSafe(t *testing.T) {var wg sync.WaitGroupvar mut sync.Mutexvar counter int = 0for i := 0; i < 5000; i++ {wg.Add(1) //每次开启一个协程,wg里面协程数量就+1go func() {defer func() {mut.Unlock()}()mut.Lock()counter++wg.Done() //每完成一个协程,wg里面的协程数量就-1}()}// time.Sleep(1 * time.Second)wg.Wait() //当wg里面所有的协程都执行完了,才会继续往下执行,这样我们//就不需要去自己估摸等待的时间了t.Log(counter)
}

三.CSP并发机制

上一小节讲解的是传统的处理并发问题的方法，而CSP并发机制则是go的一种独特的并发机制。
首先先了解一下Actor模型和CSP模型的区别,接下来我们主要介绍一下CSP机制的channel.

这种channel是最典型的一直channel机制,这种机制的的特点是:

通信双方想要完成通信必须两边都在,如果一方在另外一方不在,那么在的一方就会一直处于阻塞的状态来等待
另外一方的出现。

而上面这种channel机制我们称之为buffer channel

并不要求双方都在,而是当发送方或者接收方有需要了就去channel去放消息或者去消息。

下面我们看一下一个普通的没有使用csp的例子：

package testimport ("fmt""testing""time"
)func service() string {time.Sleep(time.Millisecond * 50) //这个服务需要持续50毫秒return "done"
}func otherTask() {fmt.Println("Working on other Task")time.Sleep(time.Millisecond * 100) //这份任务需要执行100毫秒fmt.Println("Task is done")
}func TestService(t *testing.T) {fmt.Println(service())otherTask()
}
//可以看到这里一共消耗了0.15s,说明上述任务是串行执行的,这样会导致浪费很多的时间

下面我们对Service进行包装来使用csp机制:

package testimport ("fmt""testing""time"
)func service() string {time.Sleep(time.Millisecond * 50) //这个服务需要持续50毫秒return "done"
}func otherTask() {fmt.Println("Working on other Task")time.Sleep(time.Millisecond * 100) //这份任务需要执行100毫秒fmt.Println("Task is done")
}func AsyncService() chan string {retCh := make(chan string) //这个channel只支持string类型//retCh := make(chan string,1),修改为这一行就会得到buffer channelgo func() {res := service()fmt.Println("returned service")retCh <- res //将res放入通道fmt.Println("service exited")}()return retCh
}
func TestAsyncService(t *testing.T) {retCh := AsyncService()otherTask()fmt.Println(<-retCh) //从channel取消息time.Sleep(1 * time.Second) //防止Test结束后退出导致service exited没有打印出来
}

不过细心就会发现一个问题，我们这里使用的是经典的普通channel，也就是当我们的channel的消息没有被取走的时候,那么就会阻塞在retCh <- res //将res放入通道这里，那么我们可以改为buffer channel(参加注释),

从上面我们可以看出,结果就不是必须done被打印出之后,service exited.

四.多路选择和超时控制机制

多路选择形式:

//下面只要有一个channel获取到了消息就会执行对应的语句,当有多个语句都获得了消息的,也并不会根据
//他们的顺序去执行,这个是没有保证的
select{case ret:= <-retCh1://执行语句1case ret:= <-retCh2://执行语句2...default://执行语句
}

我们可以利用多路选择来实现超时控制:

package testimport ("fmt""testing""time"
)func service() string {time.Sleep(time.Millisecond * 50) //这个服务需要持续50毫秒//time.Sleep(time.Millisecond * 200) //对应第二张结果return "done"
}func AsyncService() chan string {retCh := make(chan string, 1) //这个channel只支持string类型go func() {res := service()fmt.Println("returned service")retCh <- res //将res放入通道fmt.Println("service exited")}()return retCh
}
func TestAsyncService(t *testing.T) {select {case ret := <-AsyncService():t.Logf("%s ", ret)case <-time.After(100 * time.Millisecond):t.Error("time out")}
}

service 50ms未超时

service 200ms超时