Java并发工具CountDownLatch的使用和原理

Bertha 。 2024-03-22 22:16 74阅读 0赞

### 1.等待多线程完成的 CountDownLatch ###

CountDownLatch 允许一个或多个线程等待其他线程完成操作。

假如有这样一个需求：我们需要解析一个 Excel 里多个 sheet 的数据，此时可以考虑使用多线程，每个线程解析一个 sheet 里的数据，等到所有的 sheet 都解析完之后，程序需要提示解析完成。在这个需求中，要实现主线程等待所有线程完成 sheet 的解析操作，最简单的做法是使用 join()方法，如代码清单 1-1 所示。

**代码清单1-1 CountDownLatchUseCase.java**

public class CountDownLatchUseCase {
        public static void main(String[] args) throws InterruptedException {
            Thread threadA = new Thread(()->{
                System.out.println("parser1 finish");
            },"threadA");
    
            Thread threadB = new Thread(()->{
                System.out.println("parser2 finish");
            },"threadB");
    
            // 任务A B 开始执行
            threadA.start();
            threadB.start();
    
            // 等待AB 完成任务
            threadA.join();
            threadB.join();
        }
    }

join 用于让当前执行线程等待 join 线程执行结束。其实现原理是不停检查 join 线程是否存活，如果 join 线程存活则让当前线程永远等待。其中，wait（0）表示永远等待下去，代码片段如下。

**java.lang.Thread\#join(long)**

// 同步方法
    public final synchronized void join(long millis)
    throws InterruptedException {
    	long base = System.currentTimeMillis();
    	long now = 0;
    
    	if (millis < 0) {
    		throw new IllegalArgumentException("timeout value is negative");
    	}
    	if (millis == 0) {
            // isAlive()方法返回一个boolean值，如果线程已经启动且尚未终止，则返回true；否则，返回false。此次的while循环为了防止被阻塞的线程被意外唤醒，走出这段循环代表线程任务已经执行完毕，线程已经终止。
    		while (isAlive()) {
                // 线程进入等待状态
    			wait(0);
    		}
    	} else {
            // 使用join(long millis)参数不为0走这部分逻辑，如果线程没有即时被唤醒，超时时间过后自己进入RUNNABLE状态。
    		while (isAlive()) {
    			long delay = millis - now;
    			if (delay <= 0) {
    				break;
    			}
    			wait(delay);
    			now = System.currentTimeMillis() - base;
    		}
    	}
    }

直到 join 线程中止后，线程的 this.notifyAll()方法会被调用，调用 notifyAll()方法是在 JVM 里实现的，所以在 JDK 里看不到。  
　　在 JDK 1.5 之后的并发包中提供的 CountDownLatch 也可以实现 join 的功能，并且比join 的功能更多，将代码1-1使用CountDownLatch改下下，如代码清单 1-2 所示。

**代码清单1-2 CountDownLatchUseCase.java**

public class CountDownLatchUseCase {
        private static CountDownLatch countDownLatch = new CountDownLatch(2);
        public static void main(String[] args) throws InterruptedException {
            Thread threadA = new Thread(()->{
                System.out.println("parser1 finish");
                countDownLatch.countDown();
            },"threadA");
    
            Thread threadB = new Thread(()->{
                System.out.println("parser2 finish");
                countDownLatch.countDown();
            },"threadB");
            threadA.start();
            threadB.start();
            // 等待任务完成
            countDownLatch.await();
        }
    }

CountDownLatch 的构造函数接收一个 int 类型的参数作为计数器，如果你想等待 N个点完成，这里就传入 N。当我们调用 CountDownLatch 的 countDown 方法时，N 就会减 1，CountDownLatch 的 await 方法会阻塞当前线程，直到 N 变成零。由于 countDown方法可以用在任何地方，所以这里说的 N 个点，可以是 N 个线程，也可以是 1 个线程里的 N 个执行步骤。用在多个线程时，只需要把这个 CountDownLatch 的引用传递到线程里即可。

如果有某个解析 sheet 的线程处理得比较慢，我们不可能让主线程一直等待，所以可以使用另外一个带指定时间的 await 方法——await（long time，TimeUnit unit），这个方法等待特定时间后，就会不再阻塞当前线程。join 也有类似的方法。

> 计数器必须大于等于 0，只是等于 0 时候，计数器就是零，调用 await 方法时不会阻塞当前线程。CountDownLatch 不可能重新初始化或者修改 CountDownLatch对象的内部计数器的值。一个线程调用 countDown 方法 happen-before，另外一个线程调用 await 方法。

### 2.CountDownLatch原理解析 ###

#### 2.1 构造方法 public CountDownLatch(int count) ####

初始化好AQS同步状态为state

// 1.方法 CountDownLatch countDownLatch = new CountDownLatch(2);
    # java.util.concurrent.CountDownLatch#CountDownLatch
    // 使用构造方法将AQS的同步状态位初始为count
    public CountDownLatch(int count) {
    	if (count < 0) throw new IllegalArgumentException("count < 0");
    	this.sync = new Sync(count);
    }
    
    Sync(int count) {
    	setState(count);
    }

#### 2.2 方法public void countDown() ####

# 片段1 java.util.concurrent.CountDownLatch#countDown
    public void countDown() {
        // 将共享状态State的值减一
    	sync.releaseShared(1);
    }
    
    # 片段 2 java.util.concurrent.locks.AbstractQueuedSynchronizer#releaseShared
    public final boolean releaseShared(int arg) {
        // tryReleaseShared(arg)方法是CountDownLatch 重写的方法，该方法见片段3
    	if (tryReleaseShared(arg)) {
            // 当同步状态为0时会执行这个方法
    		doReleaseShared();
    		return true;
    	}
    	return false;
    }
    
    # 片段 3 java.util.concurrent.CountDownLatch.Sync#tryReleaseShared 模板方法
    protected boolean tryReleaseShared(int releases) {
    	// Decrement count; signal when transition to zero
        // 自旋+CAS 更新同步状态
    	for (;;) {
            // 获得同步状态
    		int c = getState();
            
    		if (c == 0)
    			return false;
    		int nextc = c-1;
    		if (compareAndSetState(c, nextc))
                // 如果同步更新后的值为0放回true
    			return nextc == 0;
    	}
    }
    
    # 片段 4 java.util.concurrent.locks.AbstractQueuedSynchronizer#doReleaseShared
    private void doReleaseShared() {
    /*
     * Ensure that a release propagates, even if there are other
     * in-progress acquires/releases.  This proceeds in the usual
     * way of trying to unparkSuccessor of head if it needs
     * signal. But if it does not, status is set to PROPAGATE to
     * ensure that upon release, propagation continues.
     * Additionally, we must loop in case a new node is added
     * while we are doing this. Also, unlike other uses of
     * unparkSuccessor, we need to know if CAS to reset status
     * fails, if so rechecking.
     */
    for (;;) {
        // 获得同步队列头节点
    	Node h = head;
        // 如果头节点存在且不等于尾节点（tail），进入if里
    	if (h != null && h != tail) {
            // 获得头节点的等待状态
    		int ws = h.waitStatus;
            // 判断等待状态是否为Node.SIGNAL，等待状态为SIGNAL表示需要唤醒后继节点
    		if (ws == Node.SIGNAL) {
                //使用compareAndSetWaitStatus(h, Node.SIGNAL, 0)方法来尝试将头节点的等待状态从Node.SIGNAL设置为0。如果设置成功，则调用unparkSuccessor(h)方法唤醒后继节点。如果设置失败，则继续循环重新检查情况。
    			if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
    				continue;            // loop to recheck cases
                // 唤醒后续节点 见代码片段5
    			unparkSuccessor(h);
    		}
            //如果等待状态为0，并且使用compareAndSetWaitStatus(h, 0, Node.PROPAGATE)方法尝试将头节点的等待状态从0设置为Node.PROPAGATE，则继续循环。
    		else if (ws == 0 &&
    				 !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
    			continue;                // loop on failed CAS
    	}
        //头节点不变退出循环
    	if (h == head)                   // loop if head changed
    		break;
    }
    }
    
    # 片段5 java.util.concurrent.locks.AbstractQueuedSynchronizer#unparkSuccessor
    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;
        // 如果等待状态ws小于0，尝试将等待状态清除为0，忽略修改失败和其他线程可能对等待状态的更改。
    	if (ws < 0)
    		compareAndSetWaitStatus(node, ws, 0);
    
    	/*
    	 * Thread to unpark is held in successor, which is normally
    	 * just the next node.  But if cancelled or apparently null,
    	 * traverse backwards from tail to find the actual
    	 * non-cancelled successor.
    	 */
        // 获得头节点的下一个节点
    	Node s = node.next;
    	if (s == null || s.waitStatus > 0) {
    		s = null;
            // 循环遍历从尾节点开始，直到找到非取消状态的节点或者已经遍历到给定节点node为止。
    		for (Node t = tail; t != null && t != node; t = t.prev)
    			if (t.waitStatus <= 0)
    				s = t;
    	}
    	if (s != null)
            // 对首个等待线程进行唤醒
    		LockSupport.unpark(s.thread);
    }

#### 2.3 方法countDownLatch.await() ####

# 片段1 java.util.concurrent.CountDownLatch#await()
    public void await() throws InterruptedException {
        // 见代码片段2
    	sync.acquireSharedInterruptibly(1);
    }
    # 片段2 java.util.concurrent.locks.AbstractQueuedSynchronizer#acquireSharedInterruptibly
    public final void acquireSharedInterruptibly(int arg)
    		throws InterruptedException {
    	// 判断线程是否被中断，中断就抛出中断异常	
    	if (Thread.interrupted())
    		throw new InterruptedException();
    	// 	调用tryAcquireShared(arg)获取同步状态信息。见代码片段3
    	if (tryAcquireShared(arg) < 0)
    	    // 同步状态不为0执行，doAcquireSharedInterruptibly(arg)见代码片段4
    		doAcquireSharedInterruptibly(arg);
    }
    
    # 片段3 java.util.concurrent.locks.AbstractQueuedSynchronizer#acquireSharedInterruptibly
    protected int tryAcquireShared(int acquires) {
        // 同步状态是否为0
    	return (getState() == 0) ? 1 : -1;
    }
    
    # 片段4 java.util.concurrent.locks.AbstractQueuedSynchronizer#doAcquireSharedInterruptibly
    private void doAcquireSharedInterruptibly(int arg)
    	throws InterruptedException {
        // 将一个代表当前线程的等待节点node添加到同步队列中。
    	final Node node = addWaiter(Node.SHARED);
        // 设置一个标志变量failed，用于跟踪获取许可的过程是否失败。
    	boolean failed = true;
    	try {
            // 循环
    		for (;;) {
                // 获取当前节点多的前驱节点
    			final Node p = node.predecessor();
                // 判断前驱节点是否为头节点
    			if (p == head) {
                    //见代码片段3。
    				int r = tryAcquireShared(arg);
    				if (r >= 0) {
                        // 同步状态为0 代表new CountDownLatch(int count) 传入的count已经被countDown()减少完了，代表任务都执行完了。
                        //设置头节点为当前节点node，并触发传播操作，将共享许可传播给后继节点。
    					setHeadAndPropagate(node, r);
    					p.next = null; // help GC
    					failed = false;
    					return;
    				}
    			}
                // shouldParkAfterFailedAcquire(p, node) 方法判断同步状态不为0时是否需要阻塞线程等待，返回true时继续执行parkAndCheckInterrupt()方法阻塞当前调用await()的线程，直到有线程调用countDown()将同步状态state更新为0时，调用 unparkSuccessor(Node node)方法将当前线程唤醒。
    			if (shouldParkAfterFailedAcquire(p, node) &&
    				parkAndCheckInterrupt())
    				throw new InterruptedException();
    		}
    	} finally {
    		if (failed)
    			cancelAcquire(node);
    	}
    }

### 3.总结 ###

CountDownLatch使用主要围绕CountDownLatch.countDown()方法和CountDownLatch.await()方法，使用时先通过CountDownLatch的构造方法设置共享同步变量state的大小，然后让其他“工作”线程完成工作任务时调用countDown方法将state-1，让“等待”线程调用await()方法等待“工作”线程完成任务，当某个“工作”线程完成最后一个任务，并且将state更新为0时，这个“工作”线程会去唤醒同步队列中的第一个“等待”线程，首个“等待”线程被唤醒会“通知”其他“等待”线程，然后他们从await()方法方法，继续执行。