同步工具类CountDownLatch

朴灿烈づ我的快乐病毒、 2022-09-10 00:22 157阅读 0赞

如果一个线程要等待5个worker线程执行完才退出可以使用用CountDownLatch

//CountDownLatch中有成员变量Sync extends AbstractQueuedSynchronizer
    //这里设置AbstractQueuedSynchronizer的state = 5
    CountDownLatch latch = new CountDownLatch(5);
    Random random = new Random();
    //启动5个worker线程
    for (int i = 0; i <5 ; i++) {
        
        new Thread(()->{
        
            try {
        
                Thread.sleep(random.nextInt(2000));
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + "运行结束");
            latch.countDown();
        }).start();
    }
    //主线程阻塞等待worker执行完毕
    latch.await();
    System.out.println("程序运行结束");

线程阻塞await

public void await() throws InterruptedException {
        
        sync.acquireSharedInterruptibly(1);
    }
    
    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        
        if (Thread.interrupted())
            throw new InterruptedException();
        //根据state状态返回
        if (tryAcquireShared(arg) < 0)
        	//如果worker线程还没执行完，需要将当前线程挂起
            doAcquireSharedInterruptibly(arg);
    }
    
    protected int tryAcquireShared(int acquires) {
        
    	//state没有减到0都返回-1
        return (getState() == 0) ? 1 : -1;
    }
    
    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) {
        
                	//如果前一个节点是head说明当前线程排在队列第一位,再次获得state
                    int r = tryAcquireShared(arg);
                    //如果state>=0，说明worker都执行完毕了
                    if (r >= 0) {
        
                    	//把当前node设置为head
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                }
                //设置pnode的状态，并挂起当前线程，被唤醒后回到for循环，把自己设置成head
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
        
            if (failed)
                cancelAcquire(node);
        }
    }

worker倒数countDown

public void countDown() {
        
        sync.releaseShared(1);
    }
    
    public final boolean releaseShared(int arg) {
        
    	//释放资源,如果state=0返回true，表示所有worker都执行完countDown
        if (tryReleaseShared(arg)) {
        
        	//释放资源后唤醒阻塞线程
            doReleaseShared();
            return true;
        }
        return false;
    }
    
    protected boolean tryReleaseShared(int releases) {
        
        // Decrement count; signal when transition to zero
        for (;;) {
        
        	//获得state
            int c = getState();
            if (c == 0)
                return false;
            int nextc = c-1;
            //cas将state-1
            if (compareAndSetState(c, nextc))
                return nextc == 0;
        }
    }
    
    private void doReleaseShared() {
        
    	for (;;) {
        
    	    Node h = head;
    	    if (h != null && h != tail) {
        
    	        int ws = h.waitStatus;
    	        //获得头节点的状态如果是Node.SIGNAL则唤醒head.next线程
    	        if (ws == Node.SIGNAL) {
        
    	            if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
    	                continue;            // loop to recheck cases
                   	//唤醒阻塞线程head.next,然后被唤醒的线程在doAcquireSharedInterruptibly方法会设置自己为新的head,下面根据h == head是否继续唤醒
    	            unparkSuccessor(h);
    	        }
    	        //修复jdk1.6特殊情况下阻塞线程的bug
    	        else if (ws == 0 &&
    	                 !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
    	            continue;                // loop on failed CAS
    	    }
    	    //判断是否继续唤醒
    	    if (h == head)                   // loop if head changed
    	        break;
    	}
    }

每一次doAcquireSharedInterruptibly方法从阻塞从被唤醒的时候，都会判断自己的前一个节点是不是head然后自旋把自己设置成head，然后doReleaseShared循环把head.next唤醒，所以这里如果多个线程await阻塞的时候，doReleaseShared会一次性唤醒队列中所有阻塞的线程