Go-并发编程基础（goroutine、channel、select等）

2022-11-12 97

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： Go-并发编程基础（goroutine、channel、select等）

概念

并发：指宏观上在一段时间内能同时运行多个程序，微观上交替运行。
并行：指同一时刻能运行多个指令。
进程：一段程序的执行过程，是系统进行资源分配的基本单位，一个进程至少有一个线程。
线程：操作系统能够进行运算调度的最小单位，它被包含在进程之中。

协程 goroutine

有独立的栈空间
共享程序堆空间
调度由用户控制
主线程是一个物理线程,直接作用在cpu上的，是重量级的，非常耗费cpu资源，
协程从主线程开启的，是轻量级的线程，是逻辑态，对资源消耗相对小。
Golang的协程机制是重要的特点，可以轻松的开启上万个协程。其它编程语言的并发机制是一般基于线程的，开启过多的线程，资源耗费大，这里就突显Golang在并发上的优势了

当一个程序启动时，其主函数即在一个单独的goroutine中运行，称之为main goroutine。新的goroutine会用go语句来创建。在语法上，go语句是一个普通的函数或方法调用前加上关键字go。go语句会使其语句中的函数在一个新创建的goroutine中运行。而go语句本身会迅速地完成。当主函数返回时，所有的goroutine都会直接打断，程序退出。

操作已经就绪，对应的goroutine就会重新分配到逻辑处理器上来完成操作。调度器对可以创建的逻辑处理器的数量没有限制，但语言运行时默认限制每个程序最多创建10000个线程。这个限制值可以通过调用runtime/debug包的SetMaxThreads方法来更改。如果程序试图使用更多的线程，就会崩溃。

goroutine调度-MPG模式

M(Main Thread):操作系统的主线程(是物理线程)，又称内核线程。

P(Processor):处理器，管理协程，例如协程执行需要的上下文等

G(Goroutine):go协程

举个例子：

分成两个部分来看

原来的情况是M主线程正在执行G0协程,另外有三个协程在队列等待如果G0协程阻塞,比如读取文件或者数据库等。

这时就会创建M1主线程(也可能是从已有的线程池中取出M1),并且将等待的3个协程挂到M1下开始执行,M0的主线程下的G0仍然执行文件io的读写。等到G0不阻塞了,M0会被放到空闲的主线程继续执行(从已有的线程池中取),同时G0又会被唤醒。

这样的MPG调度模式,可以既让G0执行,同时也不会让队列的其它协程一直阻塞,仍然可以并发/并行执行。

进行调度的调度器为Seched，它维护有存储空闲的M队列和空闲的P队列，可运行的G队列，自由的G队列以及调度器的一些状态信息等。

package main
import (
  "fmt"
  "runtime"
  "time"
)
func routinetest(name string){
  for i:=0;i<3;i++{
    fmt.Println("routinetest",i,":hello,",name)
    time.Sleep(100*time.Millisecond)
  }
}
func main() {
  num :=runtime.NumCPU()
  fmt.Println(num)
  runtime.GOMAXPROCS(num)
  //runtime.GOMAXPROCS(1)
  //--------使用go开启协程----------
  go routinetest("lady")
  go routinetest("killer")
  time.Sleep(time.Second)
}

可以通过runtime.GOMAXPROCS设置cpu数，这里设置成了8个。

通道Channel

相对于sync的低水平同步，使用channel可以实现高水平同步，channel是先进先出的。

数据结构

Channel 在运行时的内部表示是 runtime.hchan，该结构体中包含了用于保护成员变量的互斥锁，从某种程度上说，Channel 是一个用于同步和通信的有锁队列，使用互斥锁解决程序中可能存在的线程竞争问题是很常见的，我们能很容易地实现有锁队列。

type hchan struct {
  qcount   uint
  dataqsiz uint
  buf      unsafe.Pointer
  elemsize uint16
  closed   uint32
  elemtype *_type
  sendx    uint
  recvx    uint
  recvq    waitq
  sendq    waitq
  lock mutex
}

runtime.hchan 结构体中的五个字段 qcount、dataqsiz、buf、sendx、recv 构建底层的循环队列：

qcount — Channel 中的元素个数；
dataqsiz — Channel 中的循环队列的长度；
buf — Channel 的缓冲区数据指针；
sendx — Channel 的发送操作处理到的位置；
recvx — Channel 的接收操作处理到的位置；

除此之外，elemsize 和 elemtype 分别表示当前 Channel 能够收发的元素类型和大小；sendq 和 recvq 存储了当前 Channel 由于缓冲区空间不足而阻塞的 Goroutine 列表，这些等待队列使用双向链表 runtime.waitq 表示，链表中所有的元素都是 runtime.sudog 结构：

type waitq struct {
  first *sudog
  last  *sudog
}

runtime.sudog 表示一个在等待列表中的 Goroutine，该结构中存储了两个分别指向前后 runtime.sudog 的指针以构成链表。

声明&初始化

初始化需要使用make(t Type, size ...IntegerType) Type，size为缓存大小

var b chan int

var c = make(chan int)

var d = make(chan int,10)

b为nil，c为无缓存channel，d为有缓存channel

发送与接收

发送使用channel<-data，接收使用[var,ok]:=<-channel，当左侧没有变量接收时会直接丢弃掉数据，ok可以标识channel是否有数据，无数据是，接收变量获取到的是对应类型的零值。

对于nil的channel，发送和接收都会阻塞，所以不make的channel没有用，实际编程中channel应该都初始化

对于无缓存的channel，发送后会阻塞，直至接收

对于有缓存的channel，满了后发送会被阻塞，接收无影响

遍历和关闭

关闭后无法写入，只能读取，例如

close(d)

普通for循环

  for j := 0;j<len(c); j++{
    fmt.Println(<-c)
  }

若取的时候，没有其他goroutine写入的话，会读出一半。例如，刚开始len(c)是10个，当j为5时，len(c)也是5了，就跳出循环了。

  for j := 0;len(c)!=0; j++{
    fmt.Println(<-c)
  }

上面这种方法可以

for range

关闭后可以正常遍历，遍历也是从channel中接收值，大小会变化，例如

  for data := range d{
    fmt.Println(data)
  }

若不关闭，会一直接收数据，即使当前channel没有数据了，无goroutine写入时会block，若是在main routine中，会导致deadlock错误。

channel状态总结

单方向的channel

只发送chan<-int

只接收

var in =  make(chan <- int)
var out = make(<-chan int,3)

channel中的channel

package main
import "fmt"
type Request struct{
  num int
  result chan int
}
func result(r Request)  {
  r.result <- r.num + 1
}
func main() {
  r := Request{1,make(chan int)}
  go result(r)
  fmt.Println(<-r.result)
}

常见错误

panic: close of nil channel

关闭nil的channel

fatal error: all goroutines are asleep - deadlock!

main routine被永久阻塞，例如，接收一个空的channel，一直没有goroutine向里面放数据

panic: send on closed channel

向关闭的channel中发送数据

time与select

select是针对并发特有的控制结构。和switch很像，但每个case不是表达式而是通信，当有多个case可以时，将伪随机选择一个，所以不能依赖select来做顺序通信。

超时

func After(d Duration) <-chan Time

到达一定时间后可以从channel接收数据

package main
import (
  "fmt"
  "math/rand"
  "time"
)
func main() {
  timeout := time.After(2*time.Second)
  c := make(chan int)
  go func() {
    for {
      c<-0
      time.Sleep(time.Duration(rand.Intn(500))*time.Millisecond)
    }
  }()
  for {
    select {
    case <-c:
      fmt.Println("I'm working...")
    case <-timeout:
      fmt.Println("time out")
      return
    }
  }
}

时间间隔

func Tick(d Duration) <-chan Time

package main
import (
  "fmt"
  "math/rand"
  "time"
)
func main() {
  timeout := time.After(3*time.Second)
  timetrick := time.Tick(time.Second)
  c := make(chan int)
  go func() {
    for {
      c<-0
      time.Sleep(time.Duration(rand.Intn(500))*time.Millisecond)
    }
  }()
  for {
    select {
    case <-c:
      fmt.Println("I'm working...")
    case <-timetrick:
      fmt.Println("1 second pass")
    case <-timeout:
      fmt.Println("3 second")
      return
    }
  }
}

并发的后序内容查看：

Go-并发模式1(Basic Examples)

Go-并发模式2（Patterns）

更多Go相关内容：Go-Golang学习总结笔记

有问题请下方评论，转载请注明出处，并附有原文链接，谢谢！如有侵权，请及时联系