go并发

2023-07-07 本文已影响0人王侦

Go 从语言层面就支持并发。同时实现了自动垃圾回收机制。

协程：独立的栈空间，共享堆空间，调度由用户自己控制，本质上有点类似于用户级线程，这些用户级线程的调度也是自己实现的。

线程：一个线程上可以跑多个协程，协程是轻量级的线程。

Go在语言层面已经内置了调度和上下文切换的机制。

1.goroutine

在Go语言编程中你不需要去自己写进程、线程、协程，你的技能包里只有一个技能–goroutine，当你需要让某个任务并发执行的时候，你只需要把这个任务包装成一个函数，开启一个goroutine去执行这个函数就可以了，就是这么简单粗暴。

go 函数名( 参数列表 )

当main()结束时，所有在main()函数中启动的goroutine会一同结束。

2.GMP

GPM是Go语言运行时（runtime）层面的实现，是go语言自己实现的一套调度系统。区别于操作系统调度OS线程。

G - Goroutine，Go协程，是参与调度与执行的最小单位，里面除了存放本goroutine信息外还有与所在P的绑定等信息。
P - Processor，指的是逻辑处理器，P关联了的本地可运行G的队列(也称为LRQ)，最多可存放256个G。管理着一组goroutine的队列，P里面会存储当前goroutine运行的上下文环境（函数指针，堆栈地址及地址边界），P会对自己管理的goroutine队列做一些调度（比如把占用CPU时间较长的goroutine暂停、运行后续的goroutine等等）当自己的队列消费完了就去全局队列里取，如果全局队列里也消费完了会去其他P的队列里抢任务。
M - Machine，指的是系统级线程。是Go运行时（runtime）对操作系统内核线程的虚拟， M与内核线程一般是一一映射的关系，一个groutine最终是要放到M上执行的；

P与M一般也是一一对应的。他们关系是： P管理着一组G挂载在M上运行。当一个G长久阻塞在一个M上时，runtime会新建一个M，阻塞G所在的P会把其他的G 挂载在新建的M上。当旧的G阻塞完成或者认为其已经死掉时回收旧的M。

P的个数是通过runtime.GOMAXPROCS设定（最大256），Go1.5版本之后默认为物理线程数。在并发量大的时候会增加一些P和M，但不会太多，切换太频繁的话得不偿失。

单从线程调度讲，Go语言相比起其他语言的优势在于OS线程是由OS内核来调度的，goroutine则是由Go运行时（runtime）自己的调度器调度的，这个调度器使用一个称为m:n调度的技术（复用/调度m个goroutine到n个OS线程）。其一大特点是goroutine的调度是在用户态下完成的，不涉及内核态与用户态之间的频繁切换，包括内存的分配与释放，都是在用户态维护着一块大的内存池，不直接调用系统的malloc函数（除非内存池需要改变），成本比调度OS线程低很多。另一方面充分利用了多核的硬件资源，近似的把若干goroutine均分在物理线程上，再加上本身goroutine的超轻量，以上种种保证了go调度方面的性能。

3.Channel

Go语言的并发模型是CSP（Communicating Sequential Processes），提倡通过通信共享内存而不是通过共享内存而实现通信。

如果说goroutine是Go程序并发的执行体，channel就是它们之间的连接。channel是可以让一个goroutine发送特定值到另一个goroutine的通信机制。

Go 语言中的通道（channel）是一种特殊的类型。通道像一个传送带或者队列，总是遵循先入先出（First In First Out）的规则，保证收发数据的顺序。每一个通道都是一个具体类型的导管，也就是声明channel的时候需要为其指定元素类型。

channel是一种类型，一种引用类型。

var 变量 chan 类型

ch := make(chan int)
ch <- 10 // 把10发送到ch中
x := <- ch // 从ch中接收值并赋值给变量x
<-ch       // 从ch中接收值，忽略结果

3.1 无缓冲通道

无缓冲的通道只有在有人接收值的时候才能发送值。如果没有goroutine接收值，会发生死锁。

func recv(c chan int) {
    ret := <-c
    fmt.Println("接收成功", ret)
}
func main() {
    ch := make(chan int)
    go recv(ch) // 启用goroutine从通道接收值
    ch <- 10
    fmt.Println("发送成功")
}

3.2 有缓冲通道

只要通道的容量大于零，那么该通道就是有缓冲的通道，通道的容量表示通道中能存放元素的数量。通道满了就装不下了，就阻塞了，等到别人取走一个才能往里面放一个。

3.3 单向通道

func counter(out chan<- int) {
    for i := 0; i < 100; i++ {
        out <- i
    }
    close(out)
}

func squarer(out chan<- int, in <-chan int) {
    for i := range in {
        out <- i * i
    }
    close(out)
}
func printer(in <-chan int) {
    for i := range in {
        fmt.Println(i)
    }
}

func main() {
    ch1 := make(chan int)
    ch2 := make(chan int)
    go counter(ch1)
    go squarer(ch2, ch1)
    printer(ch2)
}

输出：
0
1
4
9
16
25
36
49
64
81

4.sync.WaitGroup

sync.WaitGroup 是 Go 语言标准库中的一个并发原语，它用于等待一组 Goroutine 完成任务。

sync.WaitGroup 的原理是基于计数器。它包含一个内部计数器，初始值为 0。你可以通过调用 Add() 方法增加计数器的值，表示有多少个 Goroutine 需要等待。当每个 Goroutine 完成任务后，可以调用 Done() 方法来减少计数器的值。当计数器的值为 0 时，表示所有 Goroutine 都已经完成任务，此时 Wait() 方法将解除阻塞，程序继续执行。

func main() {
    var wg sync.WaitGroup

    wg.Add(2) // 增加计数器的值，表示有两个 Goroutine 需要等待

    go func() {
        defer wg.Done() // 减少计数器的值

        // 第一个 Goroutine 的任务
        fmt.Println("Goroutine 1")
    }()

    go func() {
        defer wg.Done() // 减少计数器的值

        // 第二个 Goroutine 的任务
        fmt.Println("Goroutine 2")
    }()

    wg.Wait() // 阻塞等待，直到计数器的值为 0

    // 所有 Goroutine 已经完成任务
    fmt.Println("All Goroutines completed")
}

5.sync.Cond

sync.Cond 是 Go 语言标准库中的条件变量，它用于在多个 Goroutine 之间进行等待和通知的同步。

sync.Cond 的原理是基于条件变量的概念。它通过一个互斥锁（sync.Mutex）和一个条件变量（sync.Cond）配合使用来实现等待和通知的功能。

核心原理：

在需要等待和通知的 Goroutine 中，创建一个 sync.Cond 实例，并关联一个互斥锁。
在等待条件的 Goroutine 中，调用 Cond 的 Wait() 方法，它会在等待期间释放关联的互斥锁，并进入等待状态。
在其它 Goroutine 中，当满足某个条件时，可以调用 Cond 的 Signal() 或 Broadcast() 方法，它们会唤醒一个或多个等待中的 Goroutine。
被唤醒的 Goroutine 会重新获得关联的互斥锁，并继续执行。

func main() {
    var cond sync.Cond
    var mutex sync.Mutex
    queue := make([]int, 0)

    cond.L = &mutex // 关联互斥锁和条件变量

    waitGroup := sync.WaitGroup{}
    waitGroup.Add(2)

    go func() {
        defer waitGroup.Done()
        mutex.Lock()
        for len(queue) == 0 {
            fmt.Println("Goroutine 1: Waiting...")
            cond.Wait() // 等待条件满足
        }

        fmt.Println("Goroutine 1: Processing item from queue")
        item := queue[0]
        queue = queue[1:]
        mutex.Unlock()

        fmt.Println("Goroutine 1: Processed item:", item)
    }()

    go func() {
        defer waitGroup.Done()
        mutex.Lock()
        queue = append(queue, 42)
        fmt.Println("Goroutine 2: Added item to queue")
        cond.Signal() // 通知一个等待中的Goroutine
        mutex.Unlock()
    }()

    waitGroup.Wait()
    fmt.Println("All Goroutines completed")
}