深入学习semaphore
控制同时访问资源线程数
访问特定资源前,先使用acquire(1)获得许可,如果许可数量为0,该线程则一直阻塞,直到有可用许可。 访问资源后,使用release()释放许可。demo:
import java.util.concurrent.Semaphore;public class SemaphoreDemo { public static void main(String[] args) { int N = 8; //工人数 Semaphore semaphore = new Semaphore(5); //机器数目 for(int i=0;i
结果:
工人0占用一个机器在生产...工人2占用一个机器在生产...工人1占用一个机器在生产...工人3占用一个机器在生产...工人5占用一个机器在生产...工人2释放出机器工人5释放出机器工人1释放出机器工人4占用一个机器在生产...工人0释放出机器工人6占用一个机器在生产...工人7占用一个机器在生产...工人3释放出机器工人4释放出机器工人7释放出机器工人6释放出机器
构造方法:
public Semaphore(int permits) { //参数permits表示许可数目,即同时可以允许多少线程进行访问 sync = new NonfairSync(permits);}public Semaphore(int permits, boolean fair) { //这个多了一个参数fair表示是否是公平的,即等待时间越久的越先获取许可 sync = (fair)? new FairSync(permits) : new NonfairSync(permits);}
获取释放许可:
public void acquire() throws InterruptedException { } //获取一个许可public void acquire(int permits) throws InterruptedException { } //获取permits个许可public void release() { } //释放一个许可public void release(int permits) { } //释放permits个许可
如果想立即得到执行结果,可以使用下面几个方法:
public boolean tryAcquire() { }; //尝试获取一个许可,若获取成功,则立即返回true,若获取失败,则立即返回falsepublic boolean tryAcquire(long timeout, TimeUnit unit) throws InterruptedException { }; //尝试获取一个许可,若在指定的时间内获取成功,则立即返回true,否则则立即返回falsepublic boolean tryAcquire(int permits) { }; //尝试获取permits个许可,若获取成功,则立即返回true,若获取失败,则立即返回falsepublic boolean tryAcquire(int permits, long timeout, TimeUnit unit) throws InterruptedException { }; //尝试获取permits个许可,若在指定的时间内获取成功,则立即返回true,否则则立即返回false
源码分析:
获取许可:
public void acquire() throws InterruptedException { sync.acquireSharedInterruptibly(1); }
代码继续往下走:
public final void acquireSharedInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); if (tryAcquireShared(arg) < 0) doAcquireSharedInterruptibly(arg); }
调用doAcquireSharedInterruptibly()方法:
/** * Acquires in shared interruptible mode. * @param arg the acquire argument */ private void doAcquireSharedInterruptibly(int arg) throws InterruptedException { final Node node = addWaiter(Node.SHARED); boolean failed = true; try { for (;;) { final Node p = node.predecessor(); if (p == head) { int r = tryAcquireShared(arg); if (r >= 0) { setHeadAndPropagate(node, r); p.next = null; // help GC failed = false; return; } } if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } }
非公平策略
acquire实现,核心代码如下:
final int nonfairTryAcquireShared(int acquires) { for (;;) { int available = getState();//当前还有多少个资源可用 int remaining = available - acquires; //可用资源减去线程索取,如果少于0说明资源不够 进入下一轮循环,直到资源够 if (remaining < 0 || compareAndSetState(available, remaining))/CAS原子性操作 return remaining; }}
acquires值默认为1,表示尝试获取1个许可,remaining代表剩余的许可数。
如果remaining>0并且成功使用设置available,则获取成功.返回remaining
如果remaining < 0,表示目前没有剩余的许可.结束方法。当前线程进入AQS中的doAcquireSharedInterruptibly方法等待可用许可并挂起,直到被唤醒。2.release实现,核心代码如下:
protected final boolean tryReleaseShared(int releases) { for (;;) { int current = getState(); int next = current + releases; if (next < current) // overflow throw new Error("Maximum permit count exceeded"); if (compareAndSetState(current, next)) return true; }}
releases值默认为1,表示尝试释放1个许可,next 代表如果许可释放成功,可用许可的数量。
通过unsafe.compareAndSwapInt修改state的值,确保同一时刻只有一个线程可以释放成功。
许可释放成功,当前线程进入到AQS的doReleaseShared方法,唤醒队列中等待许可的线程。公平策略
acquire实现,核心代码如下:
protected int tryAcquireShared(int acquires) { for (;;) { if (hasQueuedPredecessors()) return -1; int available = getState(); int remaining = available - acquires; if (remaining < 0 || compareAndSetState(available, remaining)) return remaining; }}
在获取前先检查当前节点是否有前继节点(公平性)
- release实现,和非公平策略一样。