golang官方提供了一些非常优秀的类库,并不在官方版本中。这些类库在golang.org/x/……下面,可以在github.com中找到。
虽然golang提供了运行时内部使用的信号量然而并没有封装暴露成一个对外的信号量并发原语,原则上我们没有办法使用。像信号量这么常用功能也被放到了golang.org/x/sync这个扩展包里。
这个扩展包提供的semaphore也被命名为Weighted,你受得了吗。
一、信号量
Dijkstra在他的论文中为信号量定义了两个操作P和V。P操作(descrease、wait、acquire是减少信号量的计数值,而V操作(increase、signal、release)是增加信号量的计数值。
初始化信号量:设定初始的资源的数量。
P操作:将信号量的计数值减去1,如果新值已经为负,那么调用者会被阻塞并加入到等待队列中。否则,调用者会继续执行,并且获得一个资源。
V操作:将信号量的计数值加1,如果先前的计数值为负,就说明有等待的P操作的调用者。它会从等待队列中取出一个等待的调用者,唤醒它,让它继续执行。
一、结构
1 | type Weighted struct { |
- size 代表资源的最大数量。
- cur 代表当前占有的资源数量。
- mu
- waiters 被阻塞的调用者等待队列。
二、代码结构
1 | func NewWeighted(n int64) *Weighted |
- NewWeighted 提供了一个初始化方法,调用者通过这个方法设置并发数并得到一个信号量
- Acquire 相当于P操作,你可以一次获取多个资源,如果没有足够多的资源,调用者就会被阻塞。它的第一个参数是Context,这就意味着,你可以通过Context增加超时或者cancel的机制。如果是正常获取了资源,就返回nil;否则,就返回ctx.Err(),信号量不改变。
- TryAcquire 尝试获取n个资源,但是它不会阻塞,要么成功获取n个资源,返回true,要么一个也不获取,返回false。
- Release 相当于V操作,可以将n个资源释放,返还给信号量。
三、底层实现
1、Acquire
1 | func (s *Weighted) Acquire(ctx context.Context, n int64) error { |
- 因为P/V操作需要修改cur的值,因此方法先进行加锁操作。
- 当空闲资源数大于本次需要获取的资源数切等待队列为空的时候,则直接更改cur的值并解锁返回。
- 当本次需要获取的资源比最大限制的资源数还要大的时候,则直接解锁并把调用端挂起来,让他知道自己错了。
- 如果以上情况都没有满足,则会创建个队列并追加到信号量的等待队列后面,后解锁。
- 执行到此处说明信号量没有足够的资源数满足此次请求,因此会阻塞住。 在select中,等待ctx被cancel。或者当可用权重,会把cancel错误忽略掉。如果当前队头元素获取大资源后还有剩余资源则继续通知等待队列中的其他元素。
2、TryAcquire
1 | func (s *Weighted) TryAcquire(n int64) bool { |
尝试获取资源的时候如果不满足条件不会阻塞并进入等待队列,而是直接返回是否成功。
3、Release
1 | func (s *Weighted) Release(n int64) { |
- 释放资源的过程中如果释放后cur的值为负责直接panic,并且cur不会恢复。因此如果释放的数量非法,此信号量会不可用状态。
- 如果释放成功,会通知等待队列中的调用者。
- 如果等待队列为空则直接返回。
- 如果等待队列头部元素申请的资源大于于当前空闲的资源也会直接返回。
- 增加当前cur的值,并从等待队列中移除。最后关闭当前被移除元素的chan。