共计 6488 个字符,预计需要花费 17 分钟才能阅读完成。
信号量是并发编程中比拟常见的一种同步机制,它会放弃资源计数器始终在 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
外面因为是全局锁所以还须要再次加锁。
依据可用计数器信息,可分三种状况:
- 对于 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…