关于golang:更简的并发代码更强的并发控制

有没感觉 Go 的 sync 包不够用？有没遇到类型没有 sync/atomic 反对？

咱们一起看看 go-zero 的 syncx 包对规范库的一些增值补充。

https://github.com/tal-tech/g...

上面开始对以上库组件做别离介绍。

atomic

因为没有 泛型 反对，所以才会呈现多种类型的原子类反对。以下采纳 float64 作为例子：

func (f *AtomicFloat64) Add(val float64) float64 {    for {        old := f.Load()        nv := old + val        if f.CompareAndSwap(old, nv) {            return nv        }    }}func (f *AtomicFloat64) CompareAndSwap(old, val float64) bool {    return atomic.CompareAndSwapUint64((*uint64)(f), math.Float64bits(old), math.Float64bits(val))}func (f *AtomicFloat64) Load() float64 {    return math.Float64frombits(atomic.LoadUint64((*uint64)(f)))}func (f *AtomicFloat64) Set(val float64) {    atomic.StoreUint64((*uint64)(f), math.Float64bits(val))}

• Add(val)：如果 CAS 失败，一直for循环重试，获取 old val，并set old+val；
• CompareAndSwap(old, new)：调用底层 atomic 的 CAS；
• Load()：调用 atomic.LoadUint64 ，而后转换
• Set(val)：调用 atomic.StoreUint64

至于其余类型，开发者想本人扩大本人想要的类型，能够按照上述，基本上调用原始 atomic 操作，而后转换为须要的类型，比方：遇到 bool 能够借助 0, 1 来分辨对应的 false, true。

Barrier

这里 Barrier 只是将业务函数操作封装，作为闭包传入，外部将 lock 操作的加锁解锁自行解决了【避免开发者加锁了遗记解锁】

func (b *Barrier) Guard(fn func()) {    b.lock.Lock()    defer b.lock.Unlock()  // 本人的业务逻辑    fn()}

Cond/Limit/TimeoutLimit

这个数据结构和 Limit 一起组成了 TimeoutLimit ，这里将这3个一起讲：

func NewTimeoutLimit(n int) TimeoutLimit {    return TimeoutLimit{        limit: NewLimit(n),        cond:  NewCond(),    }}func NewLimit(n int) Limit {    return Limit{        pool: make(chan lang.PlaceholderType, n),    }}

• limit 这里是有缓冲的 channel；
• cond 是无缓冲的；

所以这里联合名字来了解：因为 Limit 是限度某一种资源的应用，所以须要事后在资源池中放入预置数量的资源；Cond 相似阀门，须要两边都筹备好，能力进行数据交换，所以应用无缓冲，同步控制。

这里咱们看看 stores/mongo 中对于 session 的治理，来了解 资源管制：

func (cs *concurrentSession) takeSession(opts ...Option) (*mgo.Session, error) {  // 选项参数注入    ...  // 看 limit 中是否还能取出资源    if err := cs.limit.Borrow(o.timeout); err != nil {        return nil, err    } else {        return cs.Copy(), nil    }}func (l TimeoutLimit) Borrow(timeout time.Duration) error {  // 1. 如果还有 limit 中还有资源，取出一个，返回    if l.TryBorrow() {        return nil    }    // 2. 如果 limit 中资源曾经用完了    var ok bool    for {    // 只有 cond 能够取出一个【无缓存，也只有 cond <- 此条能力通过】        timeout, ok = l.cond.WaitWithTimeout(timeout)    // 尝试取出一个【下面 cond 通过时，就有一个资源返回了】    // 看 `Return()`        if ok && l.TryBorrow() {            return nil        }        // 超时管制        if timeout <= 0 {            return ErrTimeout        }    }}func (l TimeoutLimit) Return() error {  // 返回去一个资源    if err := l.limit.Return(); err != nil {        return err    }    // 同步告诉另一个须要资源的协程【实现了阀门，两方替换】    l.cond.Signal()    return nil}

资源管理

同文件夹中还有 ResourceManager，从名字上相似，这里将两个组件放在一起解说。

先从构造上：

type ManagedResource struct {  // 资源    resource interface{}    lock     sync.RWMutex  // 生成资源的逻辑，由开发者本人管制    generate func() interface{}  // 比照资源    equals   func(a, b interface{}) bool}type ResourceManager struct {  // 资源：这里看得出来是 I/O，    resources   map[string]io.Closer    sharedCalls SharedCalls  // 对资源map互斥拜访    lock        sync.RWMutex}

而后来看获取资源的办法签名：

func (manager *ResourceManager) GetResource(key, create func() (io.Closer, error)) (io.Closer, error)// 获取一个资源（有就间接获取，没有生成一个）func (mr *ManagedResource) Take() interface{}// 判断这个资源是否不合乎传入的判断要求，不合乎则重置func (mr *ManagedResource) MarkBroken(resource interface{})

1. ResourceManager 应用 SharedCalls 做防反复申请，并将资源缓存在外部的 sourMap；另外传入的 create func 和 IO 操作无关，常见用在网络资源的缓存；
2. ManagedResource 缓存资源没有 map 而是繁多的 interface ，阐明只有一份，然而它提供了 Take() 和传入 generate()阐明能够让开发者自行更新 resource；

所以在用处上：

• ResourceManager：用在网络资源的治理。如：数据库连贯治理；
• ManagedResource：用在一些变动资源，能够做资源前后比照，达到更新资源。如：token 治理和验证

RefResource

这个就和 GC 中援用计数相似：

• Use() -> ref++
• Clean() -> ref--; if ref == 0 -> ref clean

func (r *RefResource) Use() error {  // 互斥拜访    r.lock.Lock()    defer r.lock.Unlock()    // 革除标记    if r.cleaned {        return ErrUseOfCleaned    }    // 援用 +1    r.ref++    return nil}

SharedCalls

一句话形容：应用SharedCalls能够使得同时多个申请只须要发动一次拿后果的调用，其余申请"不劳而获"，这种设计无效缩小了资源服务的并发压力，能够无效避免缓存击穿。

这个组件被重复利用在其余组件中，下面说的 ResourceManager。

相似当须要高频并发拜访一个资源时，就能够应用 SharedCalls 缓存。

// 当多个申请同时应用Do办法申请资源时func (g *sharedGroup) Do(key string, fn func() (interface{}, error)) (interface{}, error) {  // 先申请加锁  g.lock.Lock()  // 依据key，获取对应的call后果,并用变量c保留  if c, ok := g.calls[key]; ok {    // 拿到call当前，开释锁，此处call可能还没有理论数据，只是一个空的内存占位    g.lock.Unlock()    // 调用wg.Wait，判断是否有其余goroutine正在申请资源，如果阻塞，阐明有其余goroutine正在获取资源    c.wg.Wait()    // 当wg.Wait不再阻塞，示意资源获取曾经完结，能够间接返回后果    return c.val, c.err  }  // 没有拿到后果，则调用makeCall办法去获取资源，留神此处依然是锁住的，能够保障只有一个goroutine能够调用makecall  c := g.makeCall(key, fn)  // 返回调用后果  return c.val, c.err}

总结

不反复造轮子，始终是 go-zero 设计宗旨之一；也同时将平时业务积淀到组件中，这才是框架和组件的意义。

对于 go-zero 更多的设计和实现文章，能够继续关注咱们。欢送大家去关注和应用。

我的项目地址

https://github.com/tal-tech/go-zero

欢送应用 go-zero 并 star 反对咱们！

微信交换群

关注『微服务实际』公众号并回复 进群 获取社区群二维码。

go-zero 系列文章见『微服务实际』公众号