Go 的三種擴展原語之 — Semaphore

概述

信號量是併發編程中常見的一種同步機制，在需要控制訪問資源的進程數量時就會用到信號量，它會保證持有的計數器在 0 到初始化的權重之間波動。

每次獲取的資源都會將信號量中的計數器減去對應的數值，在釋放時重新加回來
當遇到計數器大於信號量大小時，會進入休眠等待其他線程釋放信號

Go 語言的擴展包中提供了帶權重的信號量 semaphore.Weighted，我們可以按照不同的權重管理資源的訪問，這種結構體暴露了 4 個方法：

semaphore.NewWeighted —— 用於創建新的信號量
semaphore.Weighted.Acquire —— 阻塞地獲取指定權重的資源，如果當前沒有空閒資源，會陷入休眠等待
semaphore.Weighted.TryAcquire —— 非阻塞地獲取指定權重的資源，如果當前沒有空閒資源，會直接返回 false
semaphore.Weighted.Release —— 用於釋放指定權重的資源

結構體

semaphore.NewWeighted 方法能提供傳入的大量權重創建一個指向 semaphore.Weighted 結構體的指針：

func NewWeighted(n int64) *Weighted {
    w := &Weighted{size: n}
    return w
}

type Weighted struct {
    size    int64
    cur     int64
    mu      sync.Mutex
    waiters list.list
}

semaphore.Weighted 結構體中包含一個 waiters 列表，其中存儲着等待獲取資源的 Goroutine。除此之外，它還包含當前信號量的上限以及一個計數器 cur，這個計數器的範圍就是 [0,size]

信號量中的計數器會隨着用戶對資源的訪問和釋放而改變，引入的權重概念能夠提供更細粒度的資源訪問控制，儘可能滿足常見用例。

semaphore.Weighted.Acquire

semaphore.Weighted.Acquire 方法能用於獲取指定權重的資源，其中包含 3 中情況：

當信號量中剩餘資源大於獲取的資源並且沒有等待的 Goroutine 時，會直接獲取信號量
當需要獲取的信號量大於 semaphore.Weighted 的上限時，由於不可能滿足條件，因此會直接返回錯誤
遇到其他情況時，會將當前 Goroutine 加入等待列表，並通過 select 等待調度器喚醒當前 Goroutine，Goroutine 被喚醒後會獲取信號量

func (s *Weighted) Acquire(ctx context.Context, n int64) error {
    if s.size - s.cur >= n && len(s.waiters) == 0 {
        s.cur += n
        return nil
    }
    
    ...
    
    ready := make(chan struct{})
    w := waiter{n: n, ready: ready}
    elem := s.waiters.PushBack(w)
    select {
    case <-ctx.Done():
        err := ctx.Err()
        select {
        case <-ready:
            err = nil
        default:
            s.waiters.Remove(elem)
        }
        return err
    case <-ready:
        return nil
    }
}

另一個用於獲取信號量的方法 semaphore.Weighted.TryAcquire 只會非阻塞地判斷當前信號量是否有充足的資源，如果有，會立刻返回 true，否則會返回 false：

func (s *Weighted) TryAcquire(n int64) bool {
    s.mu.Lock()
    success := s.size-s.cur >= n && len(s.waiters) == 0
    if success {
        s.cur += n
    }
    s.mu.Unlock()
    return success
}

因爲 semaphore.Weighted.TryAcquire 不會等待資源的釋放，所以可能更適用於一些對延時敏感、用戶需要立刻感知結果的場景

semaphore.Weighted.Release

當我們要釋放信號量時，semaphore.Weighted.Release 方法會從頭到尾遍歷 waiters 列表中全部的等待者，如果釋放資源後的信號量有充足的剩餘資源，就會通過 Channel 喚醒指定的 Goroutine：

func (c *Weighted) Relesae(n int64) {
    s.mu.Lock()
    s.cur -= n
    for {
        next := w.waiters.Front()
        if next == nil {
            break
        }
        w := next.Value.(waiter)
        if s.size-s.cur < w.n {
            break
        }
        s.cur += w.n
        s.waiters.Remove(next)
        close(w.ready)
    }
    s.mu.Unlock()
}

當然，也可能會出現剩餘資源無法喚醒 Channel 的情況，這時當前方法釋放鎖之後會直接返回。

通過 semaphore.Weighted.Release 的分析我們可以發現，如果一個信號量需要佔用的資源非常多，它可能會長時間無法獲取鎖，這也是 semaphore.Weighted.Acquire 引入上下文參數的原因，即爲信號量的獲取設置超時時間。

小結

帶權重的信號量確實有更多的應用場景，，這也是Go語言對外提供的唯一信號量實現，使用過程中需要注意以下幾個問題：

semaphore.Weighted.Acquire 和 semaphore.Weighted.TryAcquire 都可以適用於獲取資源，前者會阻塞獲取信號量，後者會非阻塞獲取信號量
semaphore.Weighted.Release 方法會按照先進先出的順序喚醒可以被喚醒的 Goroutine
如果一個 Goroutine 獲取了較多的資源，由於 semaphore.Weighted.Release 的釋放策略可能會等待較長時間

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/o3pz3jP2JlrwALXpX1G9cw

概述

結構體

semaphore.Weighted.Acquire

semaphore.Weighted.Release

小結

猜你喜歡