Go 有哪些并发同步原语？

Go是一门以并发编程见长的语言，它提供了一系列的同步原语方便开发者使用

原子操作

Mutex、RWMutex 等并发原语的底层实现是通过 atomic 包中的一些原子操作来实现的，原子操作是最基础的并发原语

package main

import (
    "fmt"
    "sync/atomic"
)

var opts int64 = 0

func main() {
    add(&opts, 3)
    load(&opts)
    compareAndSwap(&opts, 3, 4)
    swap(&opts, 5)
    store(&opts, 6)
}

func add(addr *int64, delta int64) {
    atomic.AddInt64(addr, delta) //加操作
    fmt.Println("add opts: ", *addr)
}

func load(addr *int64) {
    fmt.Println("load opts: ", atomic.LoadInt64(&opts))
}

func compareAndSwap(addr *int64, oldValue int64, newValue int64) {
    if atomic.CompareAndSwapInt64(addr, oldValue, newValue) {
        fmt.Println("cas opts: ", *addr)
        return
    }
}

func swap(addr *int64, newValue int64) {
    atomic.SwapInt64(addr, newValue)
    fmt.Println("swap opts: ", *addr)
}

func store(addr *int64, newValue int64) {
    atomic.StoreInt64(addr, newValue)
    fmt.Println("store opts: ", *addr)
}

Channel

channel 管道，高级同步原语，goroutine之间通信的桥梁

使用场景：消息队列、数据传递、信号通知、任务编排、锁

package main

import (
    "fmt"
    "time"
)

func main() {
    c := make(chan struct{}, 1)
    for i := 0; i < 10; i++ {
        go func() {
            c <- struct{}{}
            time.Sleep(1 * time.Second)
            fmt.Println("通过ch访问临界区")
            <-c
        }()
    }
    for {
    }
}

基本并发原语

Go 语言在 sync包中提供了用于同步的一些基本原语，这些基本原语提供了较为基础的同步功能，但是它们是一种相对原始的同步机制，在多数情况下，我们都应该使用抽象层级更高的 Channel 实现同步。

常见的并发原语如下：sync.Mutex、sync.RWMutex、sync.WaitGroup、sync.Cond、sync.Once、sync.Pool、sync.Context

sync.Mutex

sync.Mutex （互斥锁）可以限制对临界资源的访问，保证只有一个 goroutine 访问共享资源

使用场景：大量读写，比如多个 goroutine 并发更新同一个资源，像计数器

package main

import (
    "fmt"
    "sync"
)

func main() {
    // 封装好的计数器
    var counter Counter
    var wg sync.WaitGroup
    var gNum = 1000
    wg.Add(gNum)
    // 启动10个goroutine
    for i := 0; i < gNum; i++ {
        go func() {
            defer wg.Done()
            counter.Incr() // 受到锁保护的方法
        }()
    }
    wg.Wait()
    fmt.Println(counter.Count())
}

// 线程安全的计数器类型
type Counter struct {
    mu    sync.Mutex
    count uint64
}

// 加1的方法，内部使用互斥锁保护
func (c *Counter) Incr() {
    c.mu.Lock()
    c.count++
    c.mu.Unlock()
}

// 得到计数器的值，也需要锁保护
func (c *Counter) Count() uint64 {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.count
}

sync.RWMutex

sync.RWMutex （读写锁）可以限制对临界资源的访问，保证只有一个 goroutine 写共享资源，可以有多个goroutine 读共享资源

使用场景：大量并发读，少量并发写，有强烈的性能要求

package main

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

func main() {
    // 封装好的计数器
    var counter Counter
    var gNum = 1000
    // 启动10个goroutine
    for i := 0; i < gNum; i++ {
        go func() {
            counter.Count() // 受到锁保护的方法
        }()
    }
    for { // 一个writer
        counter.Incr() // 计数器写操作
        fmt.Println("incr")
        time.Sleep(time.Second)
    }
}

// 线程安全的计数器类型
type Counter struct {
    mu    sync.RWMutex
    count uint64
}

// 加1的方法，内部使用互斥锁保护
func (c *Counter) Incr() {
    c.mu.Lock()
    c.count++
    c.mu.Unlock()
}

// 得到计数器的值，也需要锁保护
func (c *Counter) Count() uint64 {
    c.mu.RLock()
    defer c.mu.RUnlock()
    return c.count
}

sync.WaitGroup

sync.WaitGroup 可以等待一组 Goroutine 的返回

使用场景：并发等待，任务编排，一个比较常见的使用场景是批量发出 RPC 或者 HTTP 请求

requests := []*Request{...}
wg := &sync.WaitGroup{}
wg.Add(len(requests))

for _, request := range requests {
    go func(r *Request) {
        defer wg.Done()
        // res, err := service.call(r)
    }(request)
}
wg.Wait()

sync.Cond

sync.Cond 可以让一组的 Goroutine 都在满足特定条件时被唤醒

使用场景：利用等待 / 通知机制实现阻塞或者唤醒

package main

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

var status int64

func main() {
    c := sync.NewCond(&sync.Mutex{})
    for i := 0; i < 10; i++ {
        go listen(c)
    }
    time.Sleep(1 * time.Second)
    go broadcast(c)
    time.Sleep(1 * time.Second)
}

func broadcast(c *sync.Cond) {
    c.L.Lock()
    atomic.StoreInt64(&status, 1)
    c.Signal()
    c.L.Unlock()
}

func listen(c *sync.Cond) {
    c.L.Lock()
    for atomic.LoadInt64(&status) != 1 {
        c.Wait()
    }
    fmt.Println("listen")
    c.L.Unlock()
}

sync.Once

sync.Once 可以保证在 Go 程序运行期间的某段代码只会执行一次

使用场景：常常用于单例对象的初始化场景

package main

import (
    "fmt"
    "sync"
)

func main() {
    o := &sync.Once{}
    for i := 0; i < 10; i++ {
        o.Do(func() {
            fmt.Println("only once")
        })
    }
}

sync.Pool

sync.Pool可以将暂时将不用的对象缓存起来，待下次需要的时候直接使用，不用再次经过内存分配，复用对象的内存，减轻 GC 的压力，提升系统的性能（频繁地分配、回收内存会给 GC 带来一定的负担，严重的时候会引起 CPU 的毛刺）

使用场景：对象池化， TCP连接池、数据库连接池、Worker Pool

package main

import (
    "fmt"
    "sync"
)

func main() {
    pool := sync.Pool{
        New: func() interface{} {
            return 0
        },
    }

    for i := 0; i < 10; i++ {
        v := pool.Get().(int)
        fmt.Println(v) // 取出来的值是put进去的，对象复用；如果是新建对象，则取出来的值为0
        pool.Put(i)
    }
}

sync.Map

sync.Map 线程安全的map

使用场景：map 并发读写

package main

import (
    "fmt"
    "sync"
)

func main() {
    var scene sync.Map
    // 将键值对保存到sync.Map
    scene.Store("1", 1)
    scene.Store("2", 2)
    scene.Store("3", 3)
    // 从sync.Map中根据键取值
    fmt.Println(scene.Load("1"))
    // 根据键删除对应的键值对
    scene.Delete("1")
    // 遍历所有sync.Map中的键值对
    scene.Range(func(k, v interface{}) bool {
        fmt.Println("iterate:", k, v)
        return true
    })
}

sync.Context

sync.Context 可以进行上下文信息传递、提供超时和取消机制、控制子 goroutine 的执行

使用场景：取消一个goroutine的执行

package main

import (
    "context"
    "fmt"
    "time"
)

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    go func() {
        defer func() {
            fmt.Println("goroutine exit")
        }()
        for {
            select {
            case <-ctx.Done():
                fmt.Println("receive cancel signal!")
                return
            default:
                fmt.Println("default")
                time.Sleep(time.Second)
            }
        }
    }()
    time.Sleep(time.Second)
    cancel()
    time.Sleep(2 * time.Second)
}

扩展并发原语

ErrGroup

errgroup 可以在一组 Goroutine 中提供了同步、错误传播以及上下文取消的功能

使用场景：只要一个 goroutine 出错我们就不再等其他 goroutine 了，减少资源浪费，并且返回错误

package main

import (
    "fmt"
    "net/http"

    "golang.org/x/sync/errgroup"
)

func main() {
    var g errgroup.Group
    var urls = []string{
        "http://www.baidu.com/",
        "https://www.sina.com.cn/",
    }
    for i := range urls {
        url := urls[i]
        g.Go(func() error {
            resp, err := http.Get(url)
            if err == nil {
                resp.Body.Close()
            }
            return err
        })
    }
    err := g.Wait()
    if err == nil {
        fmt.Println("Successfully fetched all URLs.")
    } else {
        fmt.Println("fetched error:", err.Error())
    }
}

Semaphore

Semaphore带权重的信号量，控制多个goroutine同时访问资源

使用场景：控制 goroutine 的阻塞和唤醒

package main

import (
    "context"
    "fmt"
    "log"
    "runtime"
    "time"

    "golang.org/x/sync/semaphore"
)

var (
    maxWorkers = runtime.GOMAXPROCS(0)
    sema       = semaphore.NewWeighted(int64(maxWorkers)) //信号量
    task       = make([]int, maxWorkers*4)

// 任务数，是worker的四
)

func main() {
    ctx := context.Background()
    for i := range task {
        // 如果没有worker可用，会阻塞在这里，直到某个worker被释放
        if err := sema.Acquire(ctx, 1); err != nil {
            break
        }
        // 启动worker goroutine
        go func(i int) {
            defer sema.Release(1)
            time.Sleep(100 * time.Millisecond) // 模拟一个耗时操作
            task[i] = i + 1
        }(i)
    }
    // 请求所有的worker,这样能确保前面的worker都执行完
    if err := sema.Acquire(ctx, int64(maxWorkers)); err != nil {
        log.Printf("获取所有的worker失败: %v", err)
    }
    fmt.Println(maxWorkers, task)
}

SingleFlight

用于抑制对下游的重复请求

使用场景：访问缓存、数据库等场景，缓存过期时只有一个请求去更新数据库

package main

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

    "golang.org/x/sync/singleflight"
)

// 模拟从数据库读取
func getArticle(id int) (article string, err error) {
    // 假设这里会对数据库进行调用, 模拟不同并发下耗时不同
    atomic.AddInt32(&count, 1)
    time.Sleep(time.Duration(count) * time.Millisecond)

    return fmt.Sprintf("article: %d", id), nil
}

// 模拟优先读缓存，缓存不存在读取数据库，并且只有一个请求读取数据库，其它请求等待
func singleflightGetArticle(sg *singleflight.Group, id int) (string, error) {
    v, err, _ := sg.Do(fmt.Sprintf("%d", id), func() (interface{}, error) {
        return getArticle(id)
    })

    return v.(string), err
}

var count int32

func main() {
    time.AfterFunc(1*time.Second, func() {
        atomic.AddInt32(&count, -count)
    })

    var (
        wg  sync.WaitGroup
        now = time.Now()
        n   = 1000
        sg  = &singleflight.Group{}
    )

    for i := 0; i < n; i++ {
        wg.Add(1)
        go func() {
            res, _ := singleflightGetArticle(sg, 1)
            // res, _ := getArticle(1)
            if res != "article: 1" {
                panic("err")
            }
            wg.Done()
        }()
    }

    wg.Wait()
    fmt.Printf("同时发起 %d 次请求，耗时: %s", n, time.Since(now))
}

本文来自投稿，不代表本站立场，如若转载，请注明出处：

原子操作

Channel

基本并发原语

sync.Mutex

sync.RWMutex

sync.WaitGroup

sync.Cond

sync.Once

sync.Pool

sync.Map

sync.Context

扩展并发原语

ErrGroup

Semaphore

SingleFlight

随机推荐

Go channel有什么特点？

channel和锁对比一下

如果一个map没申请空间，去向里面取值，会发生什么情况

Go函数参数传递到底是值传递还是引用传递？

发表评论

热门文章

标签TAG

# git

# k8s

# linux

# docker

# 渗透测试

# systemctl

# git简介

# git安装

# git配置别名

# git分支管理

# 合并冲突

# git多人协作

热评文章

最近回复

邻居

MongoDB 如何实现分片？

MongoDB 为什么选择 B-Tree 索引?

MongoDB 如何实现高可用？