CountDownLatch源码解读

我会带着你远行 2021-06-10 20:40 550阅读 0赞

#### CountDownLatch ####

本意是倒数计时，它是一个线程同步的辅助工具。它允许一个或多个线程阻塞等待直到其他线程执行完一系列的操作之后再继续执行。  
几个主要的api：

*  await()方法：阻塞等待，直到state等于0。（它是基于AbstractQueuedSynchronizer实现的）
 *  await(long timeout, TimeUnit unit)：定时等待，直到state等于0或超时。
 *  countDown()：将state减1，当state减到0之后，await阻塞的线程可以继续执行。  
    说起来还是比较简单的。

#### 使用场景 ####

当一个操作需要依赖于其他多个耗时的条件时（比如调接口获取），可以采用CountDownLatch实现并发

#### 举个例子 ####

import org.junit.Test;
    import java.util.Random;
    import java.util.concurrent.*;
    import java.util.concurrent.atomic.AtomicInteger;
    
    public class CountDownLatchTest { 
    
    	private final AtomicInteger poolNumber = new AtomicInteger(1);
    
    	/** * 定义一个线程池 */
    	private ThreadPoolExecutor executor = new ThreadPoolExecutor(20, 50, 60, TimeUnit.SECONDS,
    			new LinkedBlockingDeque<>(2000), r -> new Thread(r,"工人" + poolNumber.incrementAndGet()));
    
    	class Worker implements Runnable{ 
    
    		private CountDownLatch countDownLatch;
    
    		Worker(CountDownLatch countDownLatch) { 
    			this.countDownLatch = countDownLatch;
    		}
    
    		@Override
    		public void run() { 
    			System.out.println(Thread.currentThread().getName() + "开始执行任务");
    			try { 
    				Thread.sleep(new Random().nextInt(10000));
    			} catch (InterruptedException e) { 
    				e.printStackTrace();
    			}
    			System.out.println(Thread.currentThread().getName() + "执行任务结束");
    			countDownLatch.countDown();
    		}
    	}
    
    	@Test
    	public void testCountDownLatch() { 
    		CountDownLatch countDownLatch = new CountDownLatch(5);
    		for (int i = 0; i < 5; i++) { 
    			executor.execute(new Worker(countDownLatch));
    		}
    		try { 
    			countDownLatch.await();
    		} catch (InterruptedException e) { 
    			e.printStackTrace();
    		}
    		System.out.println("任务全部执行结束");
    	}
    }

以上示例将5个任务提交给线程池并发执行，每个子线程持有同一个countDownLatch对象，每个子线程执行结束之后执行countDown，主线程await等待。当子线程全部执行完毕之后，打印“任务全部执行结束”。

###### 执行结果：![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNzk2ODYxMw_size_16_color_FFFFFF_t_70_pic_center] ######

#### 源码 ####

以上示例比较好理解，主要问题就是countDownLatch.countDown();countDownLatch.await();两个方法，下面我们来看看他们是如何实现的。  
countDownLatch.countDown()方法：

public void countDown() { 
            sync.releaseShared(1);
        }
        
        public final boolean releaseShared(int arg) { 
        	// 释放共享信号量（如果减到0，doReleaseShared）
            if (tryReleaseShared(arg)) { 
            	// 唤醒所有阻塞的线程
                doReleaseShared();
                return true;
            }
            return false;
        }

tryReleaseShared是AbstractQueuedSynchronizer的抽象方法，在CountDownLatch的内部类Sync实现：

protected boolean tryReleaseShared(int releases) { 
                // Decrement count; signal when transition to zero
                for (;;) { 
                	// 在自旋内获取信号量
                    int c = getState();
                    // 信号量已经为0了，就不操作了，正常代码不会走到这
                    if (c == 0)
                        return false;
                    int nextc = c-1;
                      // 将减1的信号量更新
                    if (compareAndSetState(c, nextc))
                    	// 只有当信号量为0,返回true，否则false。
                        return nextc == 0;
                }
            }

当state减到0之后，执行doReleaseShared方法，唤醒因为await阻塞的线程。

private void doReleaseShared() { 
            for (;;) { 
                Node h = head;
                if (h != null && h != tail) { 
                    int ws = h.waitStatus;
                    if (ws == Node.SIGNAL) { 
                    	// 正常挂起的线程状态是SIGNAL，将waitStatus置为0，并唤醒线程
                        if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                            continue;            // loop to recheck cases
                        unparkSuccessor(h);
                    }
                    // 本来就是0（无状态）的话，则置为无条件传播PROPAGATE
                    else if (ws == 0 &&
                             !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                        continue;                // loop on failed CAS
                }
                if (h == head)                   // loop if head changed
                    break;
            }
        }

再看countDownLatch.await()：

public void await() throws InterruptedException { 
            sync.acquireSharedInterruptibly(1);
        }
        // AbstractQueuedSynchronizer
        public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException { 
    	    if (Thread.interrupted())
    	        throw new InterruptedException();
    	     
    	    if (tryAcquireShared(arg) < 0)
    	        doAcquireSharedInterruptibly(arg);
       }

tryAcquireShared也是AbstractQueuedSynchronizer的抽象方法，在CountDownLatch的内部类Sync实现：

protected int tryAcquireShared(int acquires) { 
                // 只要state不等于0，就会阻塞线程
                return (getState() == 0) ? 1 : -1;
            }

阻塞线程doAcquireSharedInterruptibly：

private void doAcquireSharedInterruptibly(int arg)
            throws InterruptedException { 
            // 创建阻塞线程节点
            final Node node = addWaiter(Node.SHARED);
            boolean failed = true;
            try { 
                for (;;) { 
                	// 如果node的前驱阶段是阻塞队列的头结点，再尝试一次获取信号量
                    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;
                        }
                    }
                    // 线程挂起前判断
                   // 1、如果线程已经是SIGNAL，但还没拿到信号量，就能够安全挂起
                   // 2、如果任务被取消CANCELLED，则移除该节点
                   // 3、否则在挂起前再将状态置为SIGNAL，再尝试一次获取信号量
                    if (shouldParkAfterFailedAcquire(p, node) &&
                    	// 挂起线程
                        parkAndCheckInterrupt())
                        throw new InterruptedException();
                }
            } finally { 
                if (failed)
                    cancelAcquire(node);
            }
        }

#### 小结 ####

1、await()：当state不等于0则尝试挂起线程  
2、countDown()：当state等于0时，则唤醒阻塞线程

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNzk2ODYxMw_size_16_color_FFFFFF_t_70_pic_center]: /images/20210518/cb158dfd945e4bb09b41fd22df965a8b.png