Davids原理探究：Java并发包中的线程同步器（CountDownLatch、CyclicBarrier和Semaphore）

清疚 2023-02-18 01:56 10阅读 0赞

### 文章目录 ###

*   *  Java并发包中的线程同步器（CountDownLatch、CyclicBarrier和Semaphore）
     *   *  CountDownLatch
         *   *  核心方法1：countDown
             *  核心方法2：await
             *  核心方法3：getCount
         *  CyclicBarrier
         *   *  核心方法1：await
             *  核心方法2：reset
         *  Semaphore
         *   *  核心方法1：construction
             *  核心方法2：acquire
             *  核心方法3：acquireUninterruptibly
             *  核心方法4：release
         *  总结

## Java并发包中的线程同步器（CountDownLatch、CyclicBarrier和Semaphore） ##

### CountDownLatch ###

使用AQS实现，通过构造函数将计数器的值赋给AQS的state，使用AQS的state字段来表示计数器的值，一次性，不可重复使用。

public CountDownLatch(int count) {
        
        if (count < 0) throw new IllegalArgumentException("count < 0");
    	// 调用Syn（class Sync extends AbstractQueuedSynchronizer）
        this.sync = new Sync(count);
    }
    
    Sync(int count) {
        
        setState(count);
    }

#### 核心方法1：countDown ####

每执行一次countDown计数器则调用AQS的releaseShared方法CAS减1，countDown不会阻塞，只会CAS失败后重试，如果减完之后==0则唤醒等待的线程处理后续操作。

public void countDown() {
        
    	// 调用AQS
        sync.releaseShared(1);
    }
    // CountDownLatch实现方法
    protected boolean tryReleaseShared(int releases) {
        
    	// 递减计数；过渡到零时发出信号
        for (;;) {
        
            int c = getState();
            if (c == 0)
                return false;
            int nextc = c-1;
            if (compareAndSetState(c, nextc))
                return nextc == 0;
        }
    }

#### 核心方法2：await ####

1.  await方法是阻塞至所有线程都调用了countDown之后并且state == 0。
2.  被其他线程中断，抛出异常。
3.  超时返回。

// 无超时await
    public void await() throws InterruptedException {
        
        sync.acquireSharedInterruptibly(1);
    }
    
    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        
    	// 如果线程被中断则抛出中断异常
        if (Thread.interrupted())
            throw new InterruptedException();
    	// 尝试-1，等价于state != 0，则放入条件队列等待，为0则直接返回
        if (tryAcquireShared(arg) < 0)
    		// AQS Sync的方法
            doAcquireSharedInterruptibly(arg);
    }
    
    // 为正数则说明state == 0
    protected int tryAcquireShared(int acquires) {
        
        return (getState() == 0) ? 1 : -1;
    }
    
    // 超时await
    public boolean await(long timeout, TimeUnit unit)
        throws InterruptedException {
        
        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    }
    
    public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)
            throws InterruptedException {
        
    	// 如果线程被中断则抛出中断异常
        if (Thread.interrupted())
            throw new InterruptedException();
    	// 如果state == 0则直接返回true，否则放入条件队列等待timeout
        return tryAcquireShared(arg) >= 0 ||
    		// AQS Sync的方法
            doAcquireSharedNanos(arg, nanosTimeout);
    }

#### 核心方法3：getCount ####

获取当前计数器的值，也就是获取AQS的state值。

// CountDownLatch类的方法
    	public long getCount() {
        
            return sync.getCount();
        }
    
    	// CountDownLatch类的方法
    	int getCount() {
        
    		// AQS Sync的方法
            return getState();
        }

### CyclicBarrier ###

回环屏障，当所有线程都调用了await方法之后，线程们就会冲破屏障，继续向下运行，可以重复使用（内部维护了两个字段，一个parties保存设置的计数器数值，一个count计算当前计数器值）。比如所有线程都执行完任务1后，再执行任务2，全部执行完任务2之后再执行任务3，声明一个CyclicBarrier即可。  
示例：

private static CyclicBarrier cyclicBarrier = new CyclicBarrier(2);
    
    public static void main( String[] args ) {
        
    
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        executorService.submit(new Thread(() -> {
        
            try {
        
                System.out.println(Thread.currentThread() + "step 1");
                cyclicBarrier.await();
                System.out.println(Thread.currentThread() + "step 2");
                cyclicBarrier.await();
                System.out.println(Thread.currentThread() + "step 3");
            } catch (InterruptedException | BrokenBarrierException e) {
        
                e.printStackTrace();
            }
        }));
        executorService.submit(new Thread(() -> {
        
            try {
        
                System.out.println(Thread.currentThread() + "step 1");
                cyclicBarrier.await();
                System.out.println(Thread.currentThread() + "step 2");
                cyclicBarrier.await();
                System.out.println(Thread.currentThread() + "step 3");
            } catch (InterruptedException | BrokenBarrierException e) {
        
                e.printStackTrace();
            }
        }));
        System.out.println("over");
        executorService.shutdown();
    }

#### 核心方法1：await ####

阻塞方法，通过独占锁ReentrantLock实现计数器原子性更新，并使用条件变量队列来实现线程同步，以下条件会返回：

1.  parties个线程调用了await方法，也就是所有线程到达屏障点（返回true）。
2.  其他线程调用了当前线程的interrupt方法中断了当前线程，则当前线程会抛出InterruptedException异常并返回。
3.  与当前屏障点关联的Generation对象的broken标志被设置为true时，会抛出BrokenBarrierException异常，然后返回。
4.  超时返回（返回false）。

调用await后计数器为0，则唤醒条件队列中的线程通过屏障执行后续操作，如果await后不为0，则将线程放入条件队列中等待。

// 非超时await
    public int await() throws InterruptedException, BrokenBarrierException {
        
        try {
        
            return dowait(false, 0L);
        } catch (TimeoutException toe) {
        
            throw new Error(toe); // cannot happen
        }
    }
    
    // 超时await
    public int await(long timeout, TimeUnit unit)
            throws InterruptedException,
                   BrokenBarrierException,
                   TimeoutException {
        
     return dowait(true, unit.toNanos(timeout));
    }
    
    private int dowait(boolean timed, long nanos)
            throws InterruptedException, BrokenBarrierException,
                   TimeoutException {
        
        // 获取可重入独占锁
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
        
        	// 查看当前屏障是否被打破，是则抛出BrokenBarrierException
            final Generation g = generation;
    
            if (g.broken)
                throw new BrokenBarrierException();
            if (Thread.interrupted()) {
        
            	// 如果当前线程被中断则:
    			// 打破屏障：generation.broken = true;
    	        // 重置计数器：count = parties;
    	        // 唤醒所有等待线程：trip.signalAll();
    	        // 并抛出中断异常InterruptedException
                breakBarrier();
                throw new InterruptedException();
            }
    		// 执行到此处说明状态正常开始执行await操作，将当前count - 1
    		// 若- 1后== 0，则将屏障设置为true并重置count
            int index = --count;
            if (index == 0) {
        
                boolean ranAction = false;
                try {
        
                    final Runnable command = barrierCommand;
                    if (command != null)
                    	// 执行任务
                        command.run();
                    ranAction = true;
                    // 激活其它因调用await方法而被阻塞的线程，重置计数器并开启新一代设置broken = false
                    nextGeneration();
                    return 0;
                } finally {
        
                    if (!ranAction)
                        breakBarrier();
                }
            }
    
            // 执行到此处说明还需等待，此处循环到中断、超时或者屏障被打破broken == true
            for (;;) {
        
                try {
        
                	// 没有超时时间
                    if (!timed)
                        trip.await();
                    // 包含超时时间
                    else if (nanos > 0L)
                        nanos = trip.awaitNanos(nanos);
                } catch (InterruptedException ie) {
        
                    if (g == generation && ! g.broken) {
        
                        breakBarrier();
                        throw ie;
                    } else {
        
                        // 即使我们没有被中断，我们也将完成等待，因此该中断被视为“属于”后续执行。
                        Thread.currentThread().interrupt();
                    }
                }
    
                if (g.broken)
                    throw new BrokenBarrierException();
    
                if (g != generation)
                    return index;
    			// 超时则打破屏障
                if (timed && nanos <= 0L) {
        
                    breakBarrier();
                    throw new TimeoutException();
                }
            }
        } finally {
        
            lock.unlock();
        }
    }
    
    // 开启新一代
    private void nextGeneration() {
        
      	// 唤醒因调用await阻塞的线程
        trip.signalAll();
        // 重置CyclicBarrier计数器
        count = parties;
        // 重置Generation，broken == false
        generation = new Generation();
    }

#### 核心方法2：reset ####

public void reset() {
        
      final ReentrantLock lock = this.lock;
        lock.lock();
        try {
        
            // 打破屏障
            breakBarrir();
            // 开启新一代
            nextGeneration();
        } finally {
        
            lock.unlock();
        }
    }
    
    // 打破屏障
    private void breakBarrier() {
        
    	// 打破屏障
      	generation.broken = true;
      	// 重置计数器
        count = parties;
        // 唤醒所有等待线程
        trip.signalAll();
    }
    
    // 开启新一代
    private void nextGeneration() {
        
    	// 唤醒所有等待线程
        trip.signalAll();
        // 重置计数器
        count = parties;
        // 重新开始新一代
        generation = new Generation();
    }

### Semaphore ###

Semaphore信号量也是Java中的一个同步器，与CountDownLatch和CyclicBarrier不同的是，它内部的计数器是递增的，并且在初始化Semaphore时可以指定一个初始值，但是并不需要知道需要同步的线程的个数，而是在需要同步的地方调用acquire方法时指定需要同步的线程个数。

#### 核心方法1：construction ####

// 默认非公平
    public Semaphore(int permits) {
        
        sync = new NonfairSync(permits);
    }
    
    // 可以指定是否公平
    public Semaphore(int permits, boolean fair) {
        
        sync = fair ? new FairSync(permits) : new NonfairSync(permits);
    }
    
    // 内部都是调用的AQS设置state状态值，根据fair构造不同的Sync
    Sync(int permits) {
        
        setState(permits);
    }

#### 核心方法2：acquire ####

// 无参方法等于acquire(1)
    public void acquire() throws InterruptedException {
        
        sync.acquireSharedInterruptibly(1);
    }
    
    // AQS类方法
    public final void acquireSharedInterruptibly(int arg)
                throws InterruptedException {
        
        // 检查线程是否被中断，中断则抛出异常
        if (Thread.interrupted())
            throw new InterruptedException();
        // 调用Semaphore实现方法tryAcquireShared
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }
    
    // NonfairSync
    protected int tryAcquireShared(int acquires) {
        
        return nonfairTryAcquireShared(acquires);
    }
    
    // 非公平锁实现
    final int nonfairTryAcquireShared(int acquires) {
        
      for (;;) {
        
      		// 获取当前信号值
            int available = getState();
            // 和参数对比，得到剩余信号量个数
            int remaining = available - acquires;
            // 如果小于0则说明还没有消耗完，直接返回负数，并执行doAcquireSharedInterruptibly(arg);进行可中断阻塞
            // 否则设置状态值为remaining并返回>= 0的remaining，退出acquire方法继续执行后续业务
            if (remaining < 0 ||
                compareAndSetState(available, remaining))
                return remaining;
        }
    }
    
    // FairSync
    protected int tryAcquireShared(int acquires) {
        
        for (;;) {
        
        	// 与非公平锁的区别在于，判断当前线程是否位于阻塞队列队首
        	// 如果队列中包含则返回负数，并执行doAcquireSharedInterruptibly(arg);进行可中断阻塞
            if (hasQueuedPredecessors())
                return -1;
            // 与非公平实现一致
            int available = getState();
            int remaining = available - acquires;
            if (remaining < 0 ||
                compareAndSetState(available, remaining))
                return remaining;
        }
    }

#### 核心方法3：acquireUninterruptibly ####

与acquire方法类似，不同之处在于该方法不受中断影响，当前线程调用acquireUninterruptibly阻塞后，如果有其它线程调用了当前线程的interrupt方法设置了当前线程的中断标志，此时当前线程并不会抛出InterruptException异常而返回。

#### 核心方法4：release ####

// 无参默认releaseShared(1)
    public void release() {
        
        sync.releaseShared(1);
    }
    
    public void release(int permits) {
        
        if (permits < 0) throw new IllegalArgumentException();
        sync.releaseShared(permits);
    }
    
    // 尝试+arg，成功则doReleaseShared，否则直接返回false
    public final boolean releaseShared(int arg) {
        
     if (tryReleaseShared(arg)) {
        
            doReleaseShared();
            return true;
        }
        return false;
    }
    
    // 循环CAS + release成功则返回true，也就是说，要么返回true要么报错，没有false的情况，-_-!!!
    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");
            if (compareAndSetState(current, next))
                return true;
        }
    }
    
    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;
                    // 如果CAS成功则唤醒队首元素
                    unparkSuccessor(h);
                }
                // CAS失败则继续循环
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;
            }
            // 如果head发生变化则继续循环，否则直接退出方法
            if (h == head)
                break;
        }
    }

### 总结 ###

线程同步器是关于线程协作的重要类，首先CountDownLatch通过计数器提供了更灵活的控制，只要检测到计数器值为0，就往下继续执行，相比于join，必须等到线程执行完毕后主线程才会继续向下运行更灵活。另外CyclicBarrier也可以达到CountDownLatch的效果，但是后者在计数器值变成0后，就不能复用了，前者则在count == 0之后会自动重置，也可以手动调用reset()方法进行重置，前者对同一算法但是输入参数不同的场景比较适用。Semaphore采用信号量递增策略，一开始并不需要关心同步线程个数，等调用acquire方法时再指定需要同步的个数，并且提供了获取信号量的公平/非公平性策略。