大型网站改版,设计网站公司选泽y湖南岚鸿询 问,咸阳做网站,免费推广软件流量精灵来看下CountDownLatch#xfffd;#xff0c;主要用于线程间通信#xff0c;await就是阻塞#xff0c;等待别人执行countDown把定义的数字减完#xff0c;就可以继续执行了#xff0c;那么去看下内部怎么实现的
CountDownLatch countDownLatch new CountDownLatch(1);
…来看下CountDownLatch主要用于线程间通信await就是阻塞等待别人执行countDown把定义的数字减完就可以继续执行了那么去看下内部怎么实现的
CountDownLatch countDownLatch new CountDownLatch(1);
countDownLatch.await();
countDownLatch.countDown();内部Sync 继承了AQS
public CountDownLatch(int count) {if (count 0) throw new IllegalArgumentException(count 0);this.sync new Sync(count);
}
private static final class Sync extends AbstractQueuedSynchronizer {private static final long serialVersionUID 4982264981922014374L;Sync(int count) {// 直接给state赋值setState(count);}int getCount() {// 现在的数量return getState();}protected int tryAcquireShared(int acquires) {// 如果等于0 返回1否则返回-1return (getState() 0) ? 1 : -1;}protected boolean tryReleaseShared(int releases) {// Decrement count; signal when transition to zerofor (;;) {// 拥有的数int c getState();// 为0的时候直接返回false不需要释放了已经都释放完了if (c 0)return false;// 减去1int nextc c - 1;// cas设置成功的话就判断是不是为0为0的时候可以后面解开阻塞if (compareAndSetState(c, nextc))return nextc 0;}}
}可以看到内部使用Sync继承了AQS然后设置的数量直接被Sync设置到了state释放的时候也不需要管传入值只能减1为0的时候直接返回获取资源的时候就不对state做改变直接判断是不是为0为0的时候返回1否则返回-1
await 方法
public void await() throws InterruptedException {sync.acquireSharedInterruptibly(1);
}
//AQS的方法
public final void acquireSharedInterruptibly(int arg)throws InterruptedException {
if (Thread.interrupted())throw new InterruptedException();
// 这里小于0就是没获取到对应资源进入队列等待
if (tryAcquireShared(arg) 0)doAcquireSharedInterruptibly(arg);
}
// 查看进入队列操作
private void doAcquireSharedInterruptibly(int arg)throws InterruptedException {// 增加共享节点这里的SHard会被存在nextWaiter
final Node node addWaiter(Node.SHARED);try {for (;;) {// 前节点final Node p node.predecessor();if (p head) {// 这里在获取这里交给子类实现的上面Sync实现的就是判断// state是不是为0为0返回1否则返回-1int r tryAcquireShared(arg);if (r 0) {// 大于0 证明state为0了等于0证明自己获取到了后续//这里放行了通知后面的setHeadAndPropagate(node, r);p.next null; // help GCreturn;}}if (shouldParkAfterFailedAcquire(p, node) parkAndCheckInterrupt())throw new InterruptedException();}} catch (Throwable t) {cancelAcquire(node);throw t;}
}private void setHeadAndPropagate(Node node, int propagate) {Node h head; // Record old head for check below// 设置其为头节点setHead(node);//这里传入的1 是大于0的就是还可以继续获取等于0的证明上个获取完了// 就需要判断如果头节点为空的或者头节点的状态if (propagate 0 || h null || h.waitStatus 0 ||(h head) null || h.waitStatus 0) {// 遍历到下一个Node s node.next;// 判断是不是共享的这里是的CountDownLatch就是用的共享锁if (s null || s.isShared())// 释放doReleaseShared();}
}// 共享的释放资源
private void doReleaseShared() {for (;;) {Node h head;// 有值if (h ! null h ! tail) {int ws h.waitStatus;if (ws Node.SIGNAL) {// signal状态设置为0成功了就直接下一步不成功一直循环if (!h.compareAndSetWaitStatus(Node.SIGNAL, 0))continue; // loop to recheck cases// unparkunparkSuccessor(h);}// 下次就为0了然后设置为传播else if (ws 0 !h.compareAndSetWaitStatus(0, Node.PROPAGATE))continue; // loop on failed CAS}if (h head) // loop if head changedbreak;}
}
// 执行了唤醒操作
private void unparkSuccessor(Node node) {/** If status is negative (i.e., possibly needing signal) try* to clear in anticipation of signalling. It is OK if this* fails or if status is changed by waiting thread.*/int ws node.waitStatus;if (ws 0)node.compareAndSetWaitStatus(ws, 0);// 这里唤醒之后Node s node.next;if (s null || s.waitStatus 0) {s null;for (Node p tail; p ! node p ! null; p p.prev)if (p.waitStatus 0)s p;}// 唤醒对应的线程让线程再次获取资源if (s ! null)LockSupport.unpark(s.thread);
}countDown 方法
public void countDown() {sync.releaseShared(1);
}
// AQS的释放共享锁的过程
public final boolean releaseShared(int arg) {// 这里的try是COuntDOwnLatch实现的是把对应的state判断下如果为0直接返回false// 那就走下面直接返回false了如果减去1之后判断是不是0相等true不等还是false// 也就是把state正好减到0走下面doReleaseSharedif (tryReleaseShared(arg)) {doReleaseShared();return true;}return false;
}
protected boolean tryReleaseShared(int releases) {// Decrement count; signal when transition to zerofor (;;) {int c getState();if (c 0)return false;int nextc c - 1;if (compareAndSetState(c, nextc))return nextc 0;}
}private void doReleaseShared() {for (;;) {Node h head;if (h ! null h ! tail) {int ws h.waitStatus;// 如果为Signalif (ws Node.SIGNAL) {// 设置为0if (!h.compareAndSetWaitStatus(Node.SIGNAL, 0))continue; // loop to recheck casesunparkSuccessor(h);}else if (ws 0 !h.compareAndSetWaitStatus(0, Node.PROPAGATE))continue; // loop on failed CAS}if (h head) // loop if head changedbreak;}
}
// 释放对应的线程
private void unparkSuccessor(Node node) {/** If status is negative (i.e., possibly needing signal) try* to clear in anticipation of signalling. It is OK if this* fails or if status is changed by waiting thread.*/int ws node.waitStatus;if (ws 0)node.compareAndSetWaitStatus(ws, 0);Node s node.next;if (s null || s.waitStatus 0) {s null;for (Node p tail; p ! node p ! null; p p.prev)if (p.waitStatus 0)s p;}if (s ! null)LockSupport.unpark(s.thread);
}总结
countDownLatch是在利用AQS的共享锁来实现的主要就是设置一个初始的值然后await是在判断state是不是为0为0的时候就返回1也就是继续向后传播会唤醒后续所有的等待节点countDown是在对设置的state减1这里countDownLatch自己实现的tryRelease方法就是对其减1然后判断结果是不是为0为0的时候返回true对排队的进行唤醒
