GO语言基础之并发concurrency

Posted 2020-10-08 爱你爱自己

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了GO语言基础之并发concurrency相关的知识，希望对你有一定的参考价值。

并发Concurrency

　　很多人都是冲着 Go 大肆宣扬的高并发而忍不住跃跃欲试，但其实从源码的解析来看，goroutine 只是由官方实现的超级“线程池”而已。不过话说回来，每个实例 4～5KB的栈内存占用和由于实现机制而大幅减少的创建和销毁开销，是制造 Go 号称的高并发的根本原因。另外，goroutine 的简单易用，也在语言层面上给予了开发者巨大的遍历。

　　高并发当中一定要注意：并发可不是并行。

　　并发主要由切换时间片来实现“同时”运行，而并行则是直接利用多核实现多线程的运行，但 Go 可以设置使用核数，以发挥多核计算机的处理能力。

　　goroutine 奉行通过通信来共享内存，而不是共享内存来通信。Go 语言主要是通过 Channe 技术通信来实现内存的共享的，因为 channel 是一个通道，Go 是通过通道来通信进行内存数据的共享。

　　对于初学者，goroutine直接理解成为线程就可以了。当对一个函数调用go，启动一个goroutine的时候，就相当于起来一个线程，执行这个函数。

　　实际上，一个goroutine并不相当于一个线程，goroutine的出现正是为了替代原来的线程概念成为最小的调度单位。一旦运行goroutine时，先去当先线程查找，如果线程阻塞了，则被分配到空闲的线程，如果没有空闲的线程，那么就会新建一个线程。注意的是，当goroutine执行完毕后，线程不会回收推出，而是成为了空闲的线程。

让我们先来看一个最简单的 goroutine 案例：

package main

import (
    "fmt"
    "time"
)

func main() {
    //启用一个goroutine
    go GoRun()
    //这里加一个休眠是因为主线程已启动就执行完毕消亡来，子线程还来不及执行
    time.Sleep(2 * time.Second)
}

func GoRun() {
    fmt.Println("Go Go Go!!!")
}

运行结果：

Go Go Go!!!

Channel

1. Channel 是 goroutine 沟通的桥梁，大都是阻塞同步的

2. 它是通过 make 创建，close 关闭

3. Channel 是引用类型

4. 可以使用 for range 来迭代，不断操作 channel

5. 可以设置单向或双向通道

6. 可以设置缓存大小，在未被填满前不会发生阻塞，即它是异步的

那么针对上溯代码我们不使用休眠，而使用 Channel 来实现我们想要的效果：

channel的意思用白话可以这么理解：主线程告诉大家你开goroutine可以，但是我在我的主线程开了一个管道，你做完了你要做的事情之后，往管道里面塞个东西告诉我你已经完成了。

package main

import (
    "fmt"
)

func main() {
    //声明创建一个通道，存储类型为bool型
    c := make(chan bool)
    //启用一个goroutine,使用的是匿名方法方式
    go func() {
        fmt.Println("Go Go Go!!!")
        c <- true  //向 channel 中存入一个值
    }()
    //当程序执行完毕之后再从通道中取出刚才赋的值
    <- c
    /**
    主线程启动了一个匿名子线程后就执行到了：<-c , 到达这里主线程就被阻塞了。只有当子线程向通道放入值后主线程阻塞才会被释放
    其实这个就是完成了消息的发送
     */
}

上溯代码可以修改为使用 for range 来进行消息的发送：

package main

import (
    "fmt"
)

func main() {
    //声明创建一个通道，存储类型为bool型，这里设置的channel就是双向通道，既可以存也可以取
    c := make(chan bool)
    //启用一个goroutine,使用的是匿名方法方式
    go func() {
        fmt.Println("Go Go Go!!!")
        c <- true  //向 channel 中存入一个值
        close(c)　 //切记如果使用for range来进行取值的时候需要在某个地方进行关闭，否则会发生死锁
    }()
    //从通道中循环取出刚才赋的值
    for v := range c {
        fmt.Println(v)
    }
}

　　从以上代码可以看出，一般使用的 Channel 都是双向通道的，即：既可以取又可以存。那单向通道一般用于什么场景下呢？

　　单向通道又分为两种，一种是只能读取，一种是只能存放，一般用于参数类型传递使用。例如有个方法返回一个Channel类型，一般要求操作只能从这里取，那么此时它的用途就是只能存放类型，如果此时你不小心存数据，此时会发生panic 导致程序奔溃发生异常。那么读取类型的Channel同理。这样做其实也是为了程序的安全性与健壮性，防止一些误操作。

　　这里还有一个知识点，就是有缓存的channel 和无缓存的channel的区别？

　　　　make(chan bool, 1) 表示带有一个缓存大小的缓存channel

　　　　make(chan bool) 或 make(chan bool, 0) 表示一个无缓存的channel

　　　　无缓存channel是阻塞的即同步的，而有缓存channel是异步的。怎么说？比如

　　　　c1:=make(chan int) 无缓冲

　　　　c2:=make(chan int,1) 有缓冲

　　　　c1 <- 1 //往无缓存通道放入数据 1

　　　　无缓冲的不仅仅是向 c1 通道放 1 而且一定要有别的线程 <- c1 接手了这个参数，那么 c1 <- 1 才会继续下去，要不然就一直阻塞着

　　　　而 c2 <- 1 则不会阻塞，因为缓冲大小是1 只有当放第二个值的时候第一个还没被人拿走，这时候才会阻塞。

　　打个比喻

　　　　无缓冲的就是一个送信人去你家门口送信，你不在家他不走，你一定要接下信，他才会走。

　　　　无缓冲保证信能到你手上

　　　　有缓冲的就是一个送信人去你家仍到你家的信箱转身就走，除非你的信箱满了他必须等信箱空下来。

　　　　有缓冲的保证信能进你家的邮箱

那如果在多线程环境下，多个线程并发抢占会使得打印不是按照顺序来，那么我们如何确保子线程全部结束完之后主线程再停止呢？主要有两种方式：

第一种：使用阻塞channel

package main

import (
    "fmt"
    "runtime"
)

func main() {
    fmt.Println("当前系统核数：", runtime.NumCPU())
    runtime.GOMAXPROCS(runtime.NumCPU()) //设置当前程序执行使用的并发数
    //定义一个阻塞channel
    c := make(chan bool)
    //这里启动10个线程运行
    for i :=0; i < 10; i++ {
        go goRun(c, i)
    }
    //我们知道一共有10次循环，那么在这里就取10次,那么子线程goRun只有都执行完了主线程取才能完毕，因为这里也循环取10次，不够的话会被阻塞
    for i := 0; i < 10; i++ {
        <- c
    }
}

func goRun(c chan bool, index int) {
    a := 1
    //循环叠加1千万次并返回最终结果
    for i := 0; i < 10000000; i++ {
        a += i
    }
    fmt.Println("线程序号:", index, a)
　　 //往阻塞队列插入内容
    c <- true
}

打印结果：

当前系统核数： 4
线程序号: 9 49999995000001
线程序号: 5 49999995000001
线程序号: 2 49999995000001
线程序号: 0 49999995000001
线程序号: 6 49999995000001
线程序号: 1 49999995000001
线程序号: 3 49999995000001
线程序号: 7 49999995000001
线程序号: 8 49999995000001
线程序号: 4 49999995000001

从打印结果可以看出，多线程环境下运行代码打印和顺序没有关系，由 CPU 调度自己决定，多运行几次打印结果一定不会一样，就是这个道理。

第二种：使用同步机制

package main

import (
    "fmt"
    "runtime"
    "sync"
)

func main() {
    fmt.Println("当前系统核数：", runtime.NumCPU())
    runtime.GOMAXPROCS(runtime.NumCPU()) //设置当前程序执行使用的并发数
    /**
    waitGroup即任务组，它的最要作用就是用来添加需要工作的任务，没完成一次任务就标记一次Done，这样任务组的待完成量会随之减1
    那么主线程就是来判断任务组内是否还有未完成任务，当没有未完成当任务之后主线程就可以结束运行，从而实现了与阻塞队列类似的同步功能
    这里创建了一个空的waitGroup(任务组)
     */
    wg := sync.WaitGroup{}
    wg.Add(10)  //添加10个任务到任务组中
    //这里启动10个线程运行
    for i :=0; i < 10; i++ {
        go goRun(&wg, i)
    }
    wg.Wait()
}

/**
这里需要传入引用类型不能传入值拷贝，因为在子线程中是需要执行Done操作，类似与我们修改结构体中的int变量主词递减，如果是只拷贝的话是不会影响原类型内的数据
这样就会发生死循环导致死锁程序奔溃，报错异常为：fatal error: all goroutines are asleep - deadlock!
 */
func goRun(wg *sync.WaitGroup, index int) {
    a := 1
    //循环叠加1千万次并返回最终结果
    for i := 0; i < 10000000; i++ {
        a += i
    }
    fmt.Println("线程序号:", index, a)

    wg.Done()
}

打印结果：

当前系统核数： 4
线程序号: 1 49999995000001
线程序号: 5 49999995000001
线程序号: 0 49999995000001
线程序号: 9 49999995000001
线程序号: 4 49999995000001
线程序号: 3 49999995000001
线程序号: 2 49999995000001
线程序号: 6 49999995000001
线程序号: 8 49999995000001
线程序号: 7 49999995000001

　　以上所有讲解到的都是基于一个 channel 来说的，那么当我们有多个 channel 的时候又该怎么处理呢？

　　Go 语言为我们提供了一种结构名为：Select，它和 switch 是非常相似的，switch 主要用于普通类型做判断的，而 select 主要是针对多个 channel 来进行判断的。

Select

1. 可处理一个或多个 channel 的发送与接收

2. 同时有多个可用的 channel 时，可以按随机顺序处理

3. 可以使用空的 select 来阻塞 main 函数

4. 它还可以设置超时时间

案例一：用多个 channel 来接收数据：

package main

import (
    "fmt"
)

/**
数据接收处理
 */
func main() {
    //批量初始化channel
    c1, c2 := make(chan int), make(chan string)
    //创建一个启动goroutine的匿名函数
    go func() {
        /**
        创建一个无限循环语句,使用select进行处理
        我们一般都是使用这种方式来处理不断的消息发送和处理
         */
        for {
            select {
            case v, ok := <- c1:
                if !ok {
                    break
                }
                fmt.Println("c1:", v)
            case v, ok := <- c2:
                if !ok {
                    break
                }
                fmt.Println("c2:", v)
            }
        }
    }()

    c1 <- 1
    c2 <- "liang"
    c1 <- 2
    c2 <- "xuli"

    //关闭channel
    close(c1)
    close(c2)
}

打印结果：

c1: 1
c2: liang
c1: 2
c2: xuli

案例二：用多个 channel 来发送数据：

package main

import (
    "fmt"
)

/**
数据接收处理，这里实现随机接收0、1 数字并打印
 */
func main() {
    c := make(chan int)
    num := 0
    //创建一个启动goroutine的匿名函数
    go func() {
        for v := range c {
            num++
            if num & 15 == 0 {
                fmt.Println()
            }
            fmt.Print(v, " ")
        }
    }()

    for {
        select {
        case c <- 0:
        case c <- 1:
        }
    }
}

打印结果：(只是粘贴了其中一部分)

1 0 1 1 0 0 0 0 1 0 1 0 0 1 0
1 0 1 1 0 1 1 0 0 1 1 1 0 0 1
1 1 1 0 0 1 1 0 0 0 0 0 1 0 1
1 1 0 0 0 1 1 1 0 0 0 1 1 0 0
1 1 0 0 0 0 0 1 0 1 1 1 1 1 1
0 1 0 0 0 0 1 0 1 1 0 1 1 1 0
1 1 0 0 0 1 1 1 0 0 0 1 0 0 1
0 1 1 1 1 0 0 1 0 0 1 1 1 1 1
1 0 0 0 0 0 1 1 1 0 1 1 0 1 1
1 0 0 0 0 1 0 0 1 0 1 0 0 1 1
0 0 1 1 1 1 1 0 0 0 1 0 0 0 1
1 1 0 1 0 1 0 1 0 0 1 1 0 0 0
1 0 0 0 1 0 0 0 1 1 0 0 0 1 1
1 0 1 1 1 1 0 0 0 1 0 0 0 1 1
1 1 0 0 1 1 0 1 0 1 0 0 0 0 1
1 1 0 0 0 1 1 0 1 0 1 0 0 0 0
0 0 1 0 0 0 1 1 1 1 1 1 1 1 0

案例三：用 channel 设置超时时间：

package main

import (
    "fmt"
    "time"
)

/**
select的超时应用
 */
func main() {
    c := make(chan bool)
    select {
    case v := <- c :
        fmt.Println(v)
    case <- time.After(3 * time.Second):
        fmt.Println("TimeOut!!!")
    }
}

打印结果：

TimeOut!!!

以上是关于GO语言基础之并发concurrency的主要内容，如果未能解决你的问题，请参考以下文章

GO语言的并发Concurrency：Goroutine与通信机制Channel

go14--并发concurrency,Goroutine ,channel

go语言设计模式之Concurrency future

go语言设计模式之Concurrency barrier

Rust编程语言入门之无畏并发

go语言设计模式之Concurrency pipeline