Go语言基础之并发！

Go语言中的并发通过goroutine实现。

goroutine类似于线程，属于用户态线程

  • 可以根据需要创建成千上万个goroutine并发工作。

goroutine是由Go语言的运行时(runtime)调度完成，而线程是由操作系统调度完成。

Go语言还提供channel在多个goroutine间进行通信。

Goroutine

Go程序会将goroutine中的任务合理地分配给每个CPU。

当需要让某个任务并发执行的时候

  • 只需要把这个任务包装成一个函数，开启一个goroutine去执行这个函数就行。

一个goroutine必定对应一个函数，可以创建多个goroutine去执行相同的函数。

启动单个goroutine

使用goroutine只需要在调用函数的时候在前面加上go关键字

  • 就可以为一个函数创建一个goroutine。
1
2
3
4
5
6
7
8
func hello(){
    fmt.Println("hello Goroutine")
}

func main(){
    go hello()
    fmt.Println("this is a main goroutine")
}

在程序启动时，Go程序就会为main()函数创建一个默认的goroutine。

当main()函数返回的时候该goroutine就结束了

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

出让资源

通过runtime.Gosched()出让资源，让其他goroutine优先执行。

1
2
3
4
5
6
7
8
9
10
11
12
func main() {
    go func(){
        for i := 0; i < 5; i++{
            fmt.Println("go")
        }
    }()
    for i:=0;i<2;i++{
        // 让出时间片，让别的goroutine先执行
        runtime.Gosched()
        fmt.Println("hello")
    }
}

自杀

通过runtime.Goexit()实现自杀，自杀前会执行提前定义的defer语句

  • 同时调用它的goroutine也会跟着自杀。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
func test(){
    // 遗嘱：临终前说的话
    defer fmt.Println("这是test的遗嘱")
    // 自杀，触发提前执行遗嘱，暴毙，后面的日志不好过了，调用它的goroutine也暴毙
    runtime.Goexit()
    // 自杀了，后面的日子不好过了
    fmt.Println("生活承诺的很多美好事情。。。(不会打印)")
}

func wildMan(){
    for i:=0;i<6;i++{
        fmt.Println("我是野人，我不喜欢约束，我讨厌制约我的主goroutine")
        time.Sleep(time.Second)
    }
}

func main() {
    // 一个会暴毙的goroutine
    go func(){
        fmt.Println("这里包含一个会暴毙的goroutine")
        test()  // runtime.Goexit()
        fmt.Println("这句应该不能出现")
    }()

    // 一个讨厌主goroutine的野人goroutine,主goroutine结束后，会把它一起带走
    go wildMan()
    for i:=0;i<=3;i++{
        time.Sleep(time.Second)
    }
}

主goroutine结束后，会带走未结束的子goroutine。

同时如果主goroutine暴毙，会令所有的子goroutine失去牵制，等所有的子goroutine都结束后

程序会崩溃：

  • fatal error: no goroutines (main called runtime.Goexit) - deadlock!。

启动多个Goroutine

使用sync.WaitGroup来实现goroutine的同步。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
var wg sync.WaitGroup

func test(){
    defer wg.Done()
    // 遗嘱：临终前说的话
    defer fmt.Println("这是test的遗嘱")
    // 自杀，触发提前执行遗嘱，暴毙，后面的日志不好过了，调用它的goroutine也暴毙
    runtime.Goexit()
    // 自杀了，后面的日子不好过了
    fmt.Println("生活承诺的很多美好事情。。。(不会打印)")
}

func wildMan(){
    defer wg.Done()
    for i:=0;i<6;i++{
        fmt.Println("我是野人，我不喜欢约束，我讨厌制约我的主goroutine")
        time.Sleep(time.Second)
    }
}

func main() {
    wg.Add(2)
    // 一个会暴毙的goroutine
    go func(){
        fmt.Println("这里包含一个会暴毙的goroutine")
        test()  // runtime.Goexit()
        fmt.Println("这句应该不能出现")
    }()

    // 一个讨厌主goroutine的野人goroutine,主goroutine结束后，会把它一起带走
    go wildMan()
    for i:=0;i<=3;i++{
        time.Sleep(time.Second)
    }
    //runtime.Goexit()
    fmt.Println("主goroutine正常退出，会带走所有的子goroutine")
    wg.Wait()  // 等待所有登记的goroutine都结束
}

Channel

channe1可以让一个goroutine发送特定值到另一个goroutine的通信机制。

Go语言中的通道(channel)是一种特殊类型，通道像一个传送带或者队列

  • 总是遵循先进先出的规则，保证数据的收发顺序

每一个通道都是一个具体类型的导管，也即是声明channel时候需要为其制定元素类型。

channel类型

channel是一种类型，一种应用类型，声明通道类型的格式如下：

1
2
3
4
5
var 变量 chan 元素类型

var ch1 chan int  // 声明一个传递整型的通道
var ch2 chan bool  // 声明一个传递布尔型的通道
var ch3 chan []int // 声明一个传递int切片的通道

创建channel

通道是引用类型，通道类型的控制是nil

1
2
var ch chan int
fmt.Println(ch)  // <nil>

声明的通道需要使用make函数初始化后才能使用

创建channel的格式如下：

1
make(chan 元素类型，[缓冲大小])

channel操作

通道有读、写和关闭三种操作。

读和写都是用<-符号。

1
2
3
4
5
6
7
8
9
10
11
12
// 初始化一个channel
ch := make(chan int)

// write to channel
ch <- 123

// read from channel
x := <- ch
<- ch  // 忽略结果

// close channel
chose(ch)

channel类型

channel分不带缓冲区的channel和带缓冲区的channel。

无缓冲区

无缓冲channel从无缓冲的channel中读取消息会阻塞

  • 直到有goroutine向该channel中发送消息。

同理，向无缓冲区的channel中发送消息也会阻塞

  • 直到有goroutine从channel中读取消息。

使用无缓冲通道进行通道将导致发送和接收的goroutine同步化

  • 因此无缓冲通道也被称为同步通道。
1
2
3
4
5
6
7
8
9
10
11
func recv(c chan int){
    ret := <-c
    fmt.Println("接收成功",ret)
}

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

有缓存的通道

有缓存的channel的声明方式为指定make函数的第二个参数

  • 该参数为channel缓存的容量。

有缓存的channel类似于一个阻塞队列（采用环形数组实现）。

当缓存未满时，向channel中发送消息不会阻塞

  • 当缓存满时，发送操作将会阻塞，直到有其他goroutine从中读取消息。

相应的，当channel中消息不为空是，读取消息不会出现阻塞

  • 当channel为空时，读取操作会发生阻塞，直到有goroutine向channel中写入消息。

可以通过使用内置的len()函数获取通道内元素的数量

  • 使用cap()函数获取通道的容量。
1
2
3
4
5
6
7
8
9
10
func main() {
    ch := make(chan int, 1) // 创建一个容量为1的有缓存区通道

    ch <- 10
    fmt.Println("len(ch) = ",len(ch))  // len(ch) =  1

    fmt.Println("cap(ch) = ",cap(ch))  // cap(ch) =  1

    fmt.Println("发送成功")
}

互斥锁

有时候在Go代码中可能存在多个goroutine同时操作一个资源（临界区）

  • 这种情况会发生竟态问题(数据竟态)。

互斥锁是一种常用的控制共享资源访问的方法

  • 它能够保证同时只有一个goroutine可以访问共享资源。

Go语言中使用sync包的Mutex类型来实现互斥锁。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
func main() {
    // 必须保证并发安全的数据
    type Account struct {
        money float32
    }

    var wg sync.WaitGroup
    account := Account{money: 1000}
    fmt.Println(account)
    
    //资源互斥锁（谁抢到锁，谁先访问资源，其他人阻塞等待）
    //全局就这么一把锁，谁先抢到谁操作，其他人被阻塞直到被释放
    var mt sync.Mutex
    
    wg.Add(2)
    // 银行卡取钱
    go func() {
        defer wg.Done()
        // 拿到互斥锁
        mt.Lock()
        // 加锁的访问
        fmt.Println("取钱前：", account.money)
        account.money -= 500
        time.Sleep(time.Nanosecond)
        fmt.Println("取钱后：", account.money)
        
        // 释放互斥锁
        mt.Unlock()
    }()
    go func(){
        defer wg.Done()
        // 拿到互斥锁（如果别人先抢到，则阻塞等待）
        mt.Lock()
        fmt.Println("存钱前：", account.money)
        account.money += 500
        time.Sleep(time.Nanosecond)
        fmt.Println("存钱后：", account.money)
        // 释放互斥锁
        mt.Unlock()
    }()

    wg.Wait()
}

通过信号量控制并发数

控制并发数属于常用的调度，规定并发的任务都必须现在某个监视管道中进行注册

  • 而这个监视管道的缓存能力是固定的
    • 比如说5，那么注册在该管道中的并发能力也是5。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
var sema chan int

// 该函数只允许5次并发执行
func f1(i int) int {
    sema <- 1
    <- time.After(2*time.Second)
    <- sema
    return i*i
}

// 信号量：通过控制管道的"带宽"（缓存能力）控制并发数
func main() {
    // 定义信号量为5"带宽"的管道
    sema = make(chan int, 5)
    var wg sync.WaitGroup
    for i:=0;i<100;i++{
        wg.Add(1)
        go func(index int) {
            ret := f1(index)
            fmt.Println(index,ret)
            wg.Done()
        }(i)
    }
    wg.Wait()
}