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

根据可用计数器信息，可分三种情况：

对于 TryAcquire() 就比较简单了，就是一个可用资源数量的判断，数量够用表示成功返回 true ，否则 false，此方法并不会进行阻塞，而是直接返回。

// TryAcquire acquires the semaphore with a weight of n without blocking.
// On success, returns true. On failure, returns false and leaves the semaphore unchanged.
func (s *Weighted) TryAcquire(n int64) bool {
    s.mu.Lock()
    success := s.size-s.cur >= n && s.waiters.Len() == 0
    if success {
        s.cur += n
    }
    s.mu.Unlock()
    return success
}

释放 Release

对于释放也很简单，就是将已使用资源数量（计数器）进行更新减少，并通知其它 waiters。

// Release releases the semaphore with a weight of n.
func (s *Weighted) Release(n int64) {
    s.mu.Lock()
    s.cur -= n
    if s.cur < 0 {
        s.mu.Unlock()
        panic("semaphore: released more than held")
    }
    s.notifyWaiters()
    s.mu.Unlock()
}

通知机制

通过 for 循环从链表头部开始头部依次遍历出链表中的所有waiter，并更新计数器 Weighted.cur，同时将其从链表中删除，直到遇到 空闲资源数量 < watier.n 为止。

func (s *Weighted) notifyWaiters() {
    for {
        next := s.waiters.Front()
        if next == nil {
            break // No more waiters blocked.
        }

        w := next.Value.(waiter)
        if s.size-s.cur < w.n {
            // Not enough tokens for the next waiter.  We could keep going (to try to
            // find a waiter with a smaller request), but under load that could cause
            // starvation for large requests; instead, we leave all remaining waiters
            // blocked.
            //
            // Consider a semaphore used as a read-write lock, with N tokens, N
            // readers, and one writer.  Each reader can Acquire(1) to obtain a read
            // lock.  The writer can Acquire(N) to obtain a write lock, excluding all
            // of the readers.  If we allow the readers to jump ahead in the queue,
            // the writer will starve — there is always one token available for every
            // reader.
            break
        }

        s.cur += w.n
        s.waiters.Remove(next)
        close(w.ready)
    }
}

可以看到如果一个链表里有多个等待者，其中一个等待者需要的资源（权重）比较多的时候，当前 watier 会出现长时间的阻塞（即使当前可用资源足够其它waiter执行，期间会有一些资源浪费），直到有足够的资源可以让这个等待者执行，然后继续执行它后面的等待者。

使用示例

官方文档提供了一个基于信号量的典型的“工作池”模式，见https://pkg.go.dev/golang.org/x/sync/semaphore#example-package-WorkerPool，演示了如何通过信号量控制一定数量的 goroutine 并发工作。

这是一个通过信号量实现并发对考拉兹猜想的示例，对1-32之间的数字进行计算，并打印32个符合结果的值。

package main

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

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

// Example_workerPool demonstrates how to use a semaphore to limit the number of
// goroutines working on parallel tasks.
//
// This use of a semaphore mimics a typical “worker pool” pattern, but without
// the need to explicitly shut down idle workers when the work is done.
func main() {
    ctx := context.TODO()

     // 权重值为逻辑cpu个数
    var (
        maxWorkers = runtime.GOMAXPROCS(0)
        sem        = semaphore.NewWeighted(int64(maxWorkers))
        out        = make([]int, 32)
    )

    // Compute the output using up to maxWorkers goroutines at a time.
    for i := range out {
        // When maxWorkers goroutines are in flight, Acquire blocks until one of the
        // workers finishes.
        if err := sem.Acquire(ctx, 1); err != nil {
            log.Printf("Failed to acquire semaphore: %v", err)
            break
        }

        go func(i int) {
            defer sem.Release(1)
            out[i] = collatzSteps(i + 1)
        }(i)
    }

    // 如果使用了 errgroup 原语则不需要下面这段语句
    if err := sem.Acquire(ctx, int64(maxWorkers)); err != nil {
        log.Printf("Failed to acquire semaphore: %v", err)
    }

    fmt.Println(out)

}

// collatzSteps computes the number of steps to reach 1 under the Collatz
// conjecture. (See https://en.wikipedia.org/wiki/Collatz_conjecture.)
func collatzSteps(n int) (steps int) {
    if n <= 0 {
        panic("nonpositive input")
    }

    for ; n > 1; steps++ {
        if steps < 0 {
            panic("too many steps")
        }

        if n%2 == 0 {
            n /= 2
            continue
        }

        const maxInt = int(^uint(0) >> 1)
        if n > (maxInt-1)/3 {
            panic("overflow")
        }
        n = 3*n + 1
    }

    return steps
}

上面先声明了总权重值为逻辑CPU数量，每次 for 循环都会调用一次 sem.Acquire(ctx, 1)，即表示最多每个CPU可运行一个 goroutine，如果当前权重值不足的话，其它groutine将处于阻塞状态，这里共循环32次，即阻塞数量最大为 32-maxWorkers。

每获取成功一个权重就会执行go匿名函数，并在函数结束时释放权重。为了保证每次for循环都会正常结束，最后调用了 sem.Acquire(ctx, int64(maxWorkers)) ，表示最后一次执行必须获取的权重值为 maxWorkers。当然如果使用 errgroup 同步原语的话，这一步可以省略掉

以下为使用 errgroup 的方法

func main() {
    ctx := context.TODO()
    var (
        maxWorkers = runtime.GOMAXPROCS(0)
        sem        = semaphore.NewWeighted(int64(maxWorkers))
        out        = make([]int, 32)
    )

    group, _ := errgroup.WithContext(context.Background())
    for i := range out {
        if err := sem.Acquire(ctx, 1); err != nil {
            log.Printf("Failed to acquire semaphore: %v", err)
            break
        }
        group.Go(func() error {
            go func(i int) {
                defer sem.Release(1)
                out[i] = collatzSteps(i + 1)
            }(i)
            return nil
        })
    }

    // 这里会阻塞，直到所有goroutine都执行完毕
    if err := group.Wait(); err != nil {
        fmt.Println(err)
    }
    fmt.Println(out)
}

转自 https://blog.haohtml.com/arch...

本文来自：Segmentfault

感谢作者：cfanbo

查看原文：Golang并发原语之-信号量Semaphore

返回前面的内容