关于golang: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 外面因为是全局锁所以还须要再次加锁。

依据可用计数器信息，可分三种状况：

1. 对于 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…