GO语言基础之并发concurrency

Posted 爱你爱自己

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!!!")
}
技术分享

运行结果:

1
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
}
技术分享

打印结果:

1
2
3
4
5
6
7
8
9
10
11
当前系统核数: 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()
}
技术分享

打印结果:

1
2
3
4
5
6
7
8
9
10
11
当前系统核数: 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)
}
技术分享

打印结果:

1
2
3
4
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
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
1 1 0 1 1 0 0 0 0 1 0 1 0 0 1 0
0 1 0 1 1 0 1 1 0 0 1 1 1 0 0 1
1 1 1 1 0 0 1 1 0 0 0 0 0 1 0 1
0 1 1 0 0 0 1 1 1 0 0 0 1 1 0 0
1 1 1 0 0 0 0 0 1 0 1 1 1 1 1 1
0 0 1 0 0 0 0 1 0 1 1 0 1 1 1 0
1 1 1 0 0 0 1 1 1 0 0 0 1 0 0 1
1 0 1 1 1 1 0 0 1 0 0 1 1 1 1 1
1 1 0 0 0 0 0 1 1 1 0 1 1 0 1 1
1 1 0 0 0 0 1 0 0 1 0 1 0 0 1 1
0 0 0 1 1 1 1 1 0 0 0 1 0 0 0 1
0 1 1 0 1 0 1 0 1 0 0 1 1 0 0 0
0 1 0 0 0 1 0 0 0 1 1 0 0 0 1 1
1 1 0 1 1 1 1 0 0 0 1 0 0 0 1 1
0 1 1 0 0 1 1 0 1 0 1 0 0 0 0 1
0 1 1 0 0 0 1 1 0 1 0 1 0 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!!!")
    }
}
技术分享

打印结果:

1
TimeOut!!!

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

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

go14--并发concurrency,Goroutine ,channel

go语言设计模式之Concurrency future

go语言设计模式之Concurrency barrier

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

go语言设计模式之Concurrency pipeline