第十八章:Go语言并发数据同步

golang-gopher.png

1. 概述

Go程序可以通过channel在多个goroutine 之间进行数据交换保障了数据正确性,但是channel的内部实现也是各种锁.

在并发环境中可能会出现竞争状态,简称 竞态 ,就是并发程序竞争访问和修改某个共享资源,存在竞态问题的程序会引起潜在问题,在写的程序中药避免出现竞态问题 ,一般竞态问题是不容易发现的,我们需要竞态检测

2. 竞态检测

检测方法

go run -race

• -race 参数,是在开始运行对竞争问题分析

package main

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

var (
    // 定义全局变量c
    c  int
    // 定义全局变量 wg (等待组)
    wg sync.WaitGroup
)

func add(n int) {
    // 消耗一个等待组计数器
    defer wg.Done()
    for i := 0; i < 2; i++ {
        // 读取全局变量c 的值
        v := c
        // 退出当前运行的goroutine,给其他goroutine机会
        runtime.Gosched()
        v++
        // 给全局变量c赋值
        c = v
    }
}
func main() {
    // 增加2个计数器
    wg.Add(2)
    go add(1)
    go add(2)
    // 等待知道计数器为0 
    wg.Wait()
    fmt.Println("it is over", c)
}

go run -race main.go

$ go run -race main.go
==================
WARNING: DATA RACE
Read at 0x000000607318 by goroutine 7:
  main.add()
      E:/Go/src/GoNote/chapter9/demo5/main/main.go:21 +0x76

Previous write at 0x000000607318 by goroutine 6:
  main.add()
      E:/Go/src/GoNote/chapter9/demo5/main/main.go:26 +0x97

Goroutine 7 (running) created at:
  main.main()
      E:/Go/src/GoNote/chapter9/demo5/main/main.go:33 +0x90

Goroutine 6 (finished) created at:
  main.main()
      E:/Go/src/GoNote/chapter9/demo5/main/main.go:32 +0x6f
==================
it is over 4
Found 1 data race(s)
exit status 66

检测的结果是发现有1处数据竞态,同时也指出了那些行代码有导致竞态

Found 1 data race(s)

3. 解决方案

我们可以明确一点就是对于共享资源的读写操作必须是原子画的,同一时刻只有一个goroutine 对共享资源星进行读写操作

Go语言中提供了原子访问 atomic包 ,互斥锁 sync.Mutex 和 等待组 sync.WaiteGroup 等方案,这些也是保障数据正确性的方案.

3.1 原子操作 atomic 包

atomic 文档

原子函数能够以很底层的加锁机制来同步访问整型变量和指针

改写上面的代码

package main

import (
    "fmt"
    "runtime"
    "sync"
    "sync/atomic"
)

var (
    // 定义全局变量c
    c  int64
    // 定义全局变量 wg (等待组)
    wg sync.WaitGroup
)

func add(n int) {
    // 消耗一个等待组计数器
    defer wg.Done()
    for i := 0; i < 2; i++ {
         //AddInt64原子性的将val的值添加到*addr并返回新值。
         atomic.AddInt64(&c,1)
         runtime.Gosched()
    }
}
func main() {
    // 增加2个计数器
    wg.Add(2)
    go add(1)
    go add(2)
    // 等待知道计数器为0
    wg.Wait()
    fmt.Println("it is over", c)
}

atomic 包中有很多的方法,我们就是看看一些简单的使用

package main

import (
    "fmt"
    "sync"
    "sync/atomic"
    "time"
)

var (
    s  int64
    wg sync.WaitGroup
)

func DO(flag string) {
    defer wg.Done()
    for {
        fmt.Println(flag)
        time.Sleep(time.Millisecond * 200)
        // 原子操作将s的值读取到
        if atomic.LoadInt64(&s) == 1 {
            fmt.Println("goroutine is stop",flag)
            break
        }
    }
}
func main() {
    wg.Add(2)
    go DO("zhangsan")
    go DO("lisi")
    time.Sleep(time.Second)
    fmt.Println("i hope goroutine stop")
    // 原子操作将 值1 存入s中
    atomic.StoreInt64(&s,1)
    wg.Wait()
}

3.2 互斥锁 mutex

互斥锁顾名思义就是相互排斥,简单的理解就是相互排斥,保障同一个时间只能有一个goroutine对共享资源进行操作 ,在上锁和释放锁的代码上创建一个临界区

package main

import (
    "fmt"
    "sync"
)

var (
    wg   sync.WaitGroup
    num  int
)

func add(n int) {
    defer wg.Done()
    num += n
}
func main() {
    wg.Add(2)
    go add(10)
    go add(100)
    wg.Wait()
    fmt.Println("main process is over")
}

go run -race main.go

main process is over
Found 1 data race(s) // 存在竞态问题
exit status 66

我们使用互斥锁解决一下

package main

import (
    "fmt"
    "sync"
)

var (
    wg   sync.WaitGroup
    lock sync.Mutex
    num  int
)

func add(n int) {
    defer wg.Done()
    // 释放锁
    defer  lock.Unlock()
    // 加锁
    lock.Lock()
    num += n
}
func main() {
    wg.Add(2)
    go add(10)
    go add(100)
    wg.Wait()
    fmt.Println("main process is over")
}

go run -race main.go

$ go run -race main.go
main process is over  // 不存在竞态问题

3.3 等待组 syc.WaitGroup

文档中解释如下

WaitGroup用于等待一组线程的结束。父线程调用Add方法来设定应等待的线程的数量。每个被等待的线程在结束时应调用Done方法。同时，主线程里可以调用Wait方法阻塞至所有线程结束。

拥有的方法
func (wg *WaitGroup) Add(delta int)	添加指定个数的计数器
func (wg *WaitGroup) Done()	减去一个计数器
func (wg *WaitGroup) Wait()	进程阻塞直到等待组计数器为0

package main

import (
    "fmt"
    "sync"
    "time"
)

var wg sync.WaitGroup
func task1(){
    defer wg.Done()
    for i:=0;i<3;i++ {
        for c:= 'a';c<'a'+26;c++ {
            fmt.Printf("%c",c)
        }
        fmt.Println("")
    }
    fmt.Println("task1 is ok")
}
func task2(){
    defer  wg.Done()
    time.Sleep(time.Millisecond*500)
    fmt.Println("task2 is ok")
}
func task3(){
    defer wg.Done()
    for i:=0;i<3;i++ {
        for c:= 'A';c<'A'+26;c++ {
            fmt.Printf("%c",c)
        }
        fmt.Println("")
    }
    fmt.Println("task3 is ok")
}
func main(){
    start := time.Now()
    wg.Add(3)
    go task1()
    go task2()
    go task3()
    // 阻塞直到计数器为0
    wg.Wait()
    fmt.Println("go go ...")
    fmt.Println("spend time ",time.Now().Sub(start))
}

go run main.go

ABCDEFGHIJKabcLMNOPQRSTUVWXYZdefghijklmnopqrstuvwxyz
abcdefghijklmnopqrs
ABCDEFGHItuvwxyz
abcdefghijklmnopqrsJKLMtuvwxyz
task1 is ok
NOPQRSTUVWXYZ
ABCDEFGHIJKLMNOPQRSTUVWXYZ
task3 is ok
task2 is ok
go go ...
spend time  500.7564ms