Golang并发原语之-信号量Semaphore-木庄网络博客

本文摘自网络，作者，侵删。

信号量是并发编程中比较常见的一种同步机制，它会保持资源计数器一直在0-N（N表示权重值大小，在用户初始化时指定）之间。当用户获取的时候会减少一会，使用完毕后再恢复过来。当遇到请求时资源不够的情况下，将会进入休眠状态以等待其它进程释放资源。

在 Golang 官方扩展库中为我们提供了一个基于权重的信号量 semaphore 并发原语。

你可以将下面的参数 n 理解为资源权重总和，表示每次获取时的权重；也可以理解为资源数量，表示每次获取时必须一次性获取的资源数量。为了理解方便，这里直接将其理解为资源数量。

数据结构

semaphore.Weighted 结构体

type waiter struct {
    n     int64
    ready chan<- struct{} // Closed when semaphore acquired.
}

// NewWeighted creates a new weighted semaphore with the given
// maximum combined weight for concurrent access.
func NewWeighted(n int64) *Weighted {
    w := &Weighted{size: n}
    return w
}

// Weighted provides a way to bound concurrent access to a resource.
// The callers can request access with a given weight.
type Weighted struct {
    size    int64
    cur     int64
    mu      sync.Mutex
    waiters list.List
}

一个 watier 就表示一个请求，其中n表示这次请求的资源数量（权重）。

使用 NewWeighted() 函数创建一个并发访问的最大资源数，这里 n 表示资源个数。

Weighted 字段说明

size 表示最大资源数量，取走时会减少，释放时会增加
cur 计数器，记录当前已使用资源数，值范围[0 - size]
mu 锁
waiters 当前处于等待休眠的请求者goroutine，每个请求者请求的资源数量可能不一样，只有在请求时，可用资源数量不足时请求者才会进入请求链表，每个请求者表示一个goroutine

计数器 cur 会随着资源的获取和释放而变化，那么为什么要引入数量（权重）这个概念呢？

方法列表

type Weighted
- func NewWeighted(n int64) *Weighted
- func (s *Weighted) Acquire(ctx context.Context, n int64) error
- func (s *Weighted) Release(n int64)
- func (s *Weighted) TryAcquire(n int64) bool

方法

NewWighted 方法用来创建一类资源，参数 n 资源表示最大可用资源总个数;
Acquire 获取指定个数的资源，如果当前没有空闲资源可用，当前请求者goroutine将陷入休眠状态;
Release 释放资源
TryAcquire 同 Acquire 一样，但当无空闲资源将直接返回false，而不阻塞。

获取 Acquire 和 TryAcquire

对于获取资源有两种方法，分别为 Acquire() 和 TryAcquire()，两者的区别我们上面已介绍过。

在获取和释放资源前必须先加全局锁。

获取资源时根据空闲资源情况，可分为三种：

有空闲资源可用，将返回nil，表示成功
请求资源数量超出了初始化时指定的总数量，这个肯定永远也不可能执行成功的，所以直接返回 ctx.Err()
当前空闲资源数量不足，需要等待其它goroutine对资源进行释放才可以运行，这时将当前请求者goroutine放入等待队列。这里再根据情况而定，具体见 select 判断

// Acquire acquires the semaphore with a weight of n, blocking until resources
// are available or ctx is done. On success, returns nil. On failure, returns
// ctx.Err() and leaves the semaphore unchanged.
//
// If ctx is already done, Acquire may still succeed without blocking.
func (s *Weighted) Acquire(ctx context.Context, n int64) error {
    // 有可用资源，直接成功返回nil
    s.mu.Lock()
    if s.size-s.cur >= n && s.waiters.Len() == 0 {
        s.cur += n
        s.mu.Unlock()
        return nil
    }

    // 请求资源权重远远超出了设置的最大权重和，失败返回 ctx.Err()
    if n > s.size {
        // Don't make other Acquire calls block on one that's doomed to fail.
        s.mu.Unlock()
        <-ctx.Done()
        return ctx.Err()
    }

    // 有部分资源可用，将请求者放在等待队列(头部），并通过select 实现通知其它waiters
    ready := make(chan struct{})
    w := waiter{n: n, ready: ready}
    // 放入链表尾部，并返回放入的元素
    elem := s.waiters.PushBack(w)
    s.mu.Unlock()

    select {
    case <-ctx.Done():
        // 收到外面的控制信号
        err := ctx.Err()
        s.mu.Lock()
        select {
        case <-ready:
            // Acquired the semaphore after we were canceled.  Rather than trying to
            // fix up the queue, just pretend we didn't notice the cancelation.
            // 如果在用户取消之前已经获取了资源,则直接忽略这个信号，返回nil表示成功
            err = nil
        default:
            // 收到控制信息，且还没有获取到资源，就直接将原来添加的 waiter 删除
            isFront := s.waiters.Front() == elem

            // 则将其从链接删除 上面 ctx.Done()
            s.waiters.Remove(elem)

            // 如果当前元素正好位于链表最前面，且还存在可用的资源，就通知其它waiters
            if isFront && s.size > s.cur {
                s.notifyWaiters()
            }
        }
        s.mu.Unlock()
        return err

    case <-ready:
        return nil
    }
}

注意上面在select逻辑语句上面有一次加解锁的操作，在 select 里面由于是全局锁所以还需要再次加锁。

阅读剩余部分