浅谈Go并发原语

原创

海风极客

发布于 2024-02-12 18:45:41

1730

发布于 2024-02-12 18:45:41

1 什么是并发原语？

在操作系统中，往往设计一些完成特定功能的、不可中断的过程，这些不可中断的过程称为原语。

因此，并发原语就是在编程语言设计之初以及后续的扩展过程中，专门为并发设计而开发的关键词或代码片段或一部分功能，进而能够为该语言实现并发提供更好的支持。

2 Go都有哪些并发原语？

2.1 官方并发原语

比较耳熟能详的就有goroutine、sync包下的Mutex、RWMutex、Once、WaitGroup、Cond、channel、Pool、Context、Timer、atomic等等

2.2 扩展并发原语

几乎都是不耳熟能详的，如Semaphore、SingleFlight、CyclicBarrier、ReentrantLock等等

3 官方并发原语的具体概念和用法

3.0 goroutine

请移步这篇文章：《深入浅出Go并发之协程—goroutine》

3.1 Mutex

Mutex是一种互斥锁，并且不可重入。类似于Java中的Lock对象。

var (
   mutex   sync.Mutex
   value   int
)

func Take() {
   var wg sync.WaitGroup
   wg.Add(3)
   go func() {
      defer wg.Done()
      AddValue()
   }()
   go func() {
      defer wg.Done()
      AddValue()
   }()
   go func() {
      defer wg.Done()
      AddValue()
   }()
   wg.Wait()
   fmt.Println(value)
}

func AddValue() {
   mutex.Lock()
   for i := 0; i < 1000; i++ {
      time.Sleep(time.Microsecond * 5)
      value++
   }
   mutex.Unlock()
}

Mutex经常被这样使用：

package main

import (
    "fmt"
    "sync"
)

type Container struct {
    mu       sync.Mutex
    counters map[string]int
}

func (c *Container) inc(name string) {

    c.mu.Lock()
    defer c.mu.Unlock()
    c.counters[name]++
}

func main() {
    c := Container{

        counters: map[string]int{"a": 0, "b": 0},
    }

    var wg sync.WaitGroup

    doIncrement := func(name string, n int) {
        for i := 0; i < n; i++ {
            c.inc(name)
        }
        wg.Done()
    }

    wg.Add(3)
    go doIncrement("a", 10000)
    go doIncrement("a", 10000)
    go doIncrement("b", 10000)

    wg.Wait()
    fmt.Println(c.counters)
}

3.2 RWMutex

RWMutex是一种读写互斥锁，该锁可以被任意数量的读取器或单个写入器持有。其他用法与Mutex类似。

3.3 Once

Once是指只执行一个动作的对象。

函数调用函数f当且仅当函数调用Do时这是Once实例的第一次。换句话说，给定var once Once,如果once.Do(f)被多次调用，只有第一次调用会调用f，即使f在每次调用中有不同的值。

type Student struct {
   Id   int
   Name string
}

var once sync.Once

func UserOnce() {
   once.Do(func() {
      var stu *Student
      stu = new(Student)
      fmt.Println(&stu)
   })
}

func TestOnce(t *testing.T) {
	for i := 0; i < 5; i++ {
		go UserOnce()
	}
	time.Sleep(5 * time.Second)
}

3.4 WaitGroup

WaitGroup等待一组goroutines完成。主goroutines调用Add来设置等待Goroutines。然后是每一个goroutine运行并在完成时调用Done。与此同时，Wait可以用来阻塞，直到所有goroutines完成。

var (
   mutex   sync.Mutex
   value   int
)

func Take() {
   var wg sync.WaitGroup
   wg.Add(3)
   go func() {
      defer wg.Done()
      AddValue()
   }()
   go func() {
      defer wg.Done()
      AddValue()
   }()
   go func() {
      defer wg.Done()
      AddValue()
   }()
   wg.Wait()
   fmt.Println(value)
}

3.5 Cond

Cond 实现了一种条件变量，可以使用多个 Reader 等待公共资源。

每个 Cond 都会关联一个 Lock ,当修改条件或者调用 Wait 方法，必须加锁，保护 Condition。有点类似 Java 中的 Wait 和 NotifyAll。

sync.Cond 条件变量是用来协调想要共享资源的那些 goroutine, 当共享资源的状态发生变化时，可以被用来通知被互斥锁阻塞的 gorountine。

3.6 channel

请移步这篇文章：《深入浅出Go语言通道chan类型》

3.7 Pool

sync.Pool 本质用途是增加临时对象的重用率，减少 GC 负担。一个Pool 可以被多个goroutine同时使用。

var createNum int32

func createBuffer() interface{} {
   atomic.AddInt32(&createNum, 1)
   buffer := make([]byte, 1024)
   return buffer
}

func Main() {
   bufferPool := &sync.Pool{New: createBuffer,}

   workerPool := 1024 * 1024
   var wg sync.WaitGroup
   wg.Add(workerPool)

   for i := 0; i < workerPool; i++ {
      go func(){
         defer wg.Done()
         buffer := bufferPool.Get()
         _ = buffer.([]byte)
         defer bufferPool.Put(buffer)
      }()
   }
   wg.Wait()
   fmt.Printf(" %d buffer objects were create.\n",createNum)
   time.Sleep(10 * time.Second)
}

3.8 Context

请移步这篇文章：《一文搞懂Go语言Context》

3.9 atomic

go语言的atomic包实现的原子操作：

atomic中的类型Value，可用来“原子的”存储或加载任意类型的值。

func(v *Value) Load() (x interface{}): 读操作，从线程安全的v中读取上一步存放的内容

func(v *Value) Store(x interface{}): 写操作，将原始的变量x存放在atomic.Value类型中

package main

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

func main() {
    var ops uint64
    var wg sync.WaitGroup
    for i := 0; i < 50; i++ {
        wg.Add(1)
        go func() {
            for c := 0; c < 1000; c++ {
                atomic.AddUint64(&ops, 1)
            }
            wg.Done()
        }()
    }
    wg.Wait()
    fmt.Println("ops:", ops)
}