java Semaphore共享锁实现原理解析
目录
- 正文
- Semaphore内部类及继承关系
- Semaphore.acquire流程分析(以非公平锁为例)
- tryAcquireShared
- doAcquireSharedInterruptibly
- Semaphore.release流程分析(以非公平锁为例)
正文
在线程间通信方式中,我们了解到可以使用Semaphore信号量来实现线程间通信,Semaphore支持公平锁和非公平锁,Semaphore底层是通过共享锁来实现的,其支持两种构造函数,如下所示:
// 默认使用非公平锁实现 public Semaphore(int permits) { sync = new NonfairSync(permits); } public Semaphore(int permits, boolean fair) { sync = fair ? new FairSync(permits) : new NonfairSync(permits); }
Semaphore提供的常用函数如下所示:
函数名 | 说明 | 备注 |
---|---|---|
acquire | 获取锁 | / |
release | 释放锁 | / |
下面我们来看下Semaphore内部的实现原理
Semaphore内部类及继承关系
可以看出Semaphore和ReentrantLock实现原理基本一致,包含NonfairSync和FairSync两个内部类,这两个内部类的父类均为AQS,不妨大胆猜测Semaphore也是依赖AQS实现的,接下来我们一起来看下Semaphore获取和释放锁的流程。
Semaphore.acquire流程分析(以非公平锁为例)
从上图可以看出,针对阻塞线程的部分实现,和ReentrantLock基本一致,我们不做赘述,主要来看下前半部分的源码实现:
// Semaphore.Java public void acquire() throws InterruptedException { sync.acquireSharedInterruptibly(1); }
// AbstractQueuedSynchronizer.java public final void acquireSharedInterruptibly(int arg) throws InterruptedException { // 如果线程是中断状态,抛出异常 if (Thread.interrupted()) throw new InterruptedException(); // 尝试获取共享资源 if (tryAcquireShared(arg) < 0) doAcquireSharedInterruptibly(arg); }
从源码可以看出acquire主要依赖于tryAcquireShared和doAcquireSharedInterruptibly,接下来我们来分别看下这两块的代码
tryAcquireShared
// NonfairSync protected int tryAcquireShared(int acquires) { return nonfairTryAcquireShared(acquires); }
// Sync final int nonfairTryAcquireShared(int acquires) { for (;;) { int available = getState(); int remaining = available - acquires; if (remaining < 0 || compareAndSetState(available, remaining)) return remainijsng; } }
// AbstractQueuedSynchronizer.java protected final boolean compareAndSetState(int expect, int update) { // See below for intrinsics setup to support this return unsafe.compareAndSwapInt(this, stateOffset, expect, update); }
从代码可以看出这里主要是根据申请的许可证数量,比较时否有许可证数量,如果可用许可证数量小于0,则直接返回,如果大于0,则通过CAS将state设置为可用许可证数量。
doAcquireSharedInterruptibly
当tryAcquireShared中返回的可用许可证数量小于0时,执行doAcquireSharedInterruptibly流程,代码如下:
// AbstractQueuedSynchronizer.java // 在队尾新建Node对象并添加 private Node addwaiter(Node mode) { Node node = new Node(Thread.currentThread(), mode); // Try the fast pathttp://www.devze.comh of enq; backup to full enq on failure Node pred = tail; if (pred != null) { node.prev = pred; if (compareAndSetTail(pred, node)) { pred.next = node; return node; } } enq(node); return node; }
// AbstractQueuedSynchronizer.java private void doAcquireSharedInterruptibly(int arg) throws InterruptedException { // 将当前线程添加到等待队列 final Node node = addWaiter(Node.SHARED); boolean failed = true; try { // for循环自旋 for (;;) { // 获取nohttp://www.devze.comde的前一个节点 开发者_开发培训 js final Node p = node.predecessor(); // 如果前一个节点是头节点 if (p == head) { // 尝试获取锁 int r = tryAcquireShared(arg); if (r >= 0) { // 获取锁成功,更新node信息设置为头节点,并通知其他节点 setHeadAndPropagate(node, r); p.next = null; // help GC failed = false; return; } } // 判断是否需要阻塞线程,设置waitStatus并阻塞 if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckIntewww.devze.comrrupt()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); } }
private void setHeadAndPropagate(Node node, int propagate) { Node h = head; // Record old head for check below setHead(node); if (propagate > 0 || h == null || h.waitStatus < 0 || (h = head) == null || h.waitStatus < 0) { Node s = node.next; if (s == null || s.isShared()) doReleaseShared(); } }
执行setHeadAndPropagate的主要目的在于,这里能获取到说明在该线程自旋过程中有线程释放了许可证,释放的许可证数量有可能还有剩余,所以传递给其他节点的线程,唤醒其他阻塞状态的线程也尝试去获取许可证。
Semaphore.release流程分析(以非公平锁为例)
Semaphore.release流程相对而言,就比较简单,将release传递到AQS内部通过CAS更新许可证数量信息,更新完成后,遍历队列中Node节点,将Node waitStatus设置为0,并对对应线程执行unpark,相关代码如下:
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"); // 通过CAS更新许可证数量 if (compareAndSetState(current, next)) return true; } }
private void unparkSuccessor(Node node) { int ws = node.waitStatus; if (ws < 0) compareAndSetWaitStatus(node, ws, 0); Node s = node.next; if (s == null || s.waitStatus > 0) { s = null; for (Node t = tail; t != null && t != node; t = t.prev) if (t.waitStatus <= 0) s = t; } if (s != null) LockSupport.unpark(s.thread); } // 许可证数量更新完成后,调用该方法唤醒线程 private void doReleaseShared() { // 自旋 for (;;) { Node h = head; if (h != null && h != tail) { int ws = h.waitStatus; if (ws == Node.SIGNAL) { if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) continue; // loop to recheck cases // 唤醒后继节点线程抢占许可证 unparkSuccessor(h); } else if (ws == 0 && !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) continue; // loop on failed CAS } if (h == head) // loop if head changed break; } }
综上,我们分析了Smaphore非公平锁的实现,感兴趣的可以分析下公平锁的实现,其本质区别在于在tryAcquireShared中只有当等待队列为空时,才会去尝试更新剩余许可证数量。
以上就是Semaphore共享锁实现原理解析的详细内容,更多关于Semaphore共享锁的资料请关注我们其它相关文章!
精彩评论