Java之juc旅途-同步工具类(三)

「爱情、让人受尽委屈。」 2023-09-28 16:40 143阅读 0赞

### 概述 ###

一些场景下，我们需要去协调多个线程去同步进行一些逻辑，jdk提供了三种类给我们。

*  CountDownLatch：可以指定数量的线程同步运行，只执行一次。
 *  CyclicBarrier：可以指定数量的线程同步运行，可循环执行无数次。
 *  Semaphore：可以控制访问某个资源的线程数量。

### CountDownLatch ###

CountDownLatch 是基于AQS的共享模式做的(由Sync类去实现AQS类)，当 CountDownLatch 初始化时就会给锁状态 state 指定一个初始的值（可以理解为锁的重入次数）。

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) {
        
                setState(count);
            }
    
            int getCount() {
        
                return getState();
            }
    
            protected int tryAcquireShared(int acquires) {
        
                return (getState() == 0) ? 1 : -1;
            }
    
            protected boolean tryReleaseShared(int releases) {
        
                // Decrement count; signal when transition to zero
                for (;;) {
        
                    int c = getState();
                    if (c == 0)
                        return false;
                    int nextc = c - 1;
                    if (compareAndSetState(c, nextc))
                        return nextc == 0;
                }
            }
        }

调用 await 方法就会直调 AbstractQueuedSynchronizer 获取共享锁的方法 acquireSharedInterruptibly，因为初始已经是有一个初始值了，所以会被阻塞。

public void await() throws InterruptedException {
        
            sync.acquireSharedInterruptibly(1);
        }

调用 countDown 方法就会直调 AbstractQueuedSynchronizer 释放共享锁的方法 releaseShared 执行一次释放锁操作，当锁状态到达 0 则锁表示已经被完全释放了，这时候 await 阻塞的所有线程都被唤醒。

public void countDown() {
        
            sync.releaseShared(1);
        }

[  
][Link 1]

### CyclicBarrier ###

CyclicBarrier顾名思义就是一个栅栏，可以实现让指定数量线程等待至某个状态之后再全部同步执行。  
在所有等待线程都被释放之后，又是一个新的栅栏，又可以实现让指定数量线程等待至某个状态之后再全部同步执行，这样无限循环。  
他是基于ReentrantLock来实现的，先看来关键变量与构造方法。

// 重入锁主要用于保证线程安全
        private final ReentrantLock lock = new ReentrantLock();
        // Condition用于阻塞和唤醒线程
        private final Condition trip = lock.newCondition();
    	// 表示一组线程的总数量
        private final int parties;
    	// 本变量表示当所有线程到达同步点且被唤醒前 将要被执行的操作 （可以为null 表示无操作）
        private final Runnable barrierCommand;
    	// 每一轮的栅栏
        private Generation generation = new Generation();
    	// 表示还未执行到同步点的线程数量 当该变量为0时表示所有线程均已执行到同步点
    	private int count;
    	
        public CyclicBarrier(int parties) {
        
            this(parties, null);
        }

栅栏是由Generation控制的，他只有一个变量\*\*broken(布尔值)，\*\*代表着是否已经被破坏。

当某一线程到达同步点时，则调用await()方法，此过程中会使count自减，此时如果计数器值非零，说明仍有线程未到达同步点，则当前线程使用ReentrantLock的Condition的**await()进入阻塞状态。**

/**
    	* 哪个线程调用了await()方法，就表示该线程已经执行到了同步点
    	* CyclicBarrier提供了两个版本的await()方法
    	* 无论是否设定时间阈值 最终都要调用dowait()方法 核心逻辑都在此方法中
    	*/
    	public int await() throws InterruptedException, BrokenBarrierException {
        
            try {
        
                return dowait(false, 0L);
            } catch (TimeoutException toe) {
        
                throw new Error(toe); // cannot happen
            }
        }
    
        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 {
        
            /**
            * 首先线程尝试获取独占锁
            * 以此保证不会有多个线程同时修改count和generation等变量 保证了线程安全
            */
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
        
            	//获取当前代
                final Generation g = generation;
    			//如果当前代已被破坏 则抛出异常
                if (g.broken)
                    throw new BrokenBarrierException();
    			/**
    			* 对应Generation注释中的情况1 待加入等待队列的线程被中断 则判定当前代被破坏
    			* 此处breakBarrier()会将当前代的broken设为true 同时唤醒其他线程
    			* 这样如果其他被唤醒的线程再次执行await()方法
    			* 便会止步于↑↑↑前一个判断并抛出异常
    			*/
                if (Thread.interrupted()) {
        
                    breakBarrier();
                    throw new InterruptedException();
                }
    			/**
    			* index用于表示剩余未到达同步点的线程数量
    			* 至于为什么此处不直接使用count变量 笔者猜想是基于线程安全考虑
    			* 纵观dowait方法 
    			* 虽然同时只有一个线程可以持有锁 且dowait方法内除自减再无对count的写操作
    			* 但仍存在用于重置栅栏的非私有方法reset()可以并发的修改count的值 进而引发线程安全问题
    			*/
                int index = --count;
                //index为0 表示全部线程执行到同步点
                if (index == 0) {
        
                	//用于判断下列try块中代码（具体是barrierCommand）是否顺利执行
                    boolean ranAction = false;
                    try {
        
                    	//在唤醒其他线程前 若有指定 则优先执行barrierCommand
                        final Runnable command = barrierCommand;
                        if (command != null)
                            command.run();
                        //执行成功
                        ranAction = true;
                        //换代操作 
                        //该方法会唤醒等待队列中的线程 创建一个新的Generation实例 并重置count
                        nextGeneration();
                        return 0;
                    } finally {
        
                    	/*
                    	* 如果ranAction为false 表示barrierCommand执行过程中出现异常
                    	* 对应了Generation注释中的情况2
                    	*/
                    	
                        if (!ranAction)
                            breakBarrier();
                    }
                }
    
                // 若能执行到这里 代表仍有线程未到达同步点
                for (;;) {
        
                    try {
        
                    	/**
                    	* 根据是否设定了时间阈值 分别调用不同版本的await方法
                    	* await方法要做的是将当前线程加入到等待队列中（阻塞）
    					*/
                        if (!timed)
                            trip.await();
                        else if (nanos > 0L)
                            nanos = trip.awaitNanos(nanos);
                    } catch (InterruptedException ie) {
        
                    	/**
                    	* 捕获到了中断异常 仍对应Generation注释中的情况1 需要将broken表示设置为true
                    	*/
                        if (g == generation && ! g.broken) {
        
                            breakBarrier();
                            throw ie;
                        } else {
        
                        	//如果不满足if条件 说明线程不属于当前一代 中断                    	
                            Thread.currentThread().interrupt();
                        }
                    }
    				//检测到当前代已被破坏 抛出异常
                    if (g.broken)
                        throw new BrokenBarrierException();
    				//dowait执行过程中 或执行了nextGeneration或外界调用reset 已经换代 方法返回
                    if (g != generation)
                        return index;
    				/**
    				* 等待超时 对应Generation注释中的情况三 将broken设为true
    				*/
                    if (timed && nanos <= 0L) {
        
                        breakBarrier();
                        throw new TimeoutException();
                    }
                }
            } finally {
        
            	//无论执行结果如何 释放重入锁 以便同步队列中的其他线程尝试获取锁
                lock.unlock();
            }
        }

当满足条件时（比如栅栏被破坏了，或者是count为0了），就会调用ReentrantLock的Condition的**signalAll()释放被阻塞的线程，这样就达到了控制线程的效果了。**

正常情况下的话，会去构建新的栅栏，然后进入下一轮

private void nextGeneration() {
        
            // signal completion of last generation
            trip.signalAll();
            // set up next generation
            count = parties;
            generation = new Generation();
        }

### Semaphore ###

Semaphore用于控制同时访问某些资源的线程个数，具体事物做法为通过调用 acquire() 来获取一个许可，如果没有许可的话，就等待，在许可使用完后通过 release() 释放该许可，以便其他线程继续使用。  
跟CountDownLatch一样，他也是基于AQS的共享模式做的(由Sync类去实现AQS类)，初始化时就会给锁状态 state 指定一个初始的值（可以理解为锁的重入次数）,只不过他具有公平和非公平两种模式。

// 默认不公平
    public Semaphore(int permits) {
        
      sync = new NonfairSync(permits);
    }
    
    // 指定是否公平
    public Semaphore(int permits, boolean fair) {
        
      sync = fair ? new FairSync(permits) : new NonfairSync(permits);
    }

不公平：

static final class NonfairSync extends Sync {
        
            private static final long serialVersionUID = -2694183684443567898L;
    
            NonfairSync(int permits) {
        
                super(permits);
            }
    
            protected int tryAcquireShared(int acquires) {
        
                return nonfairTryAcquireShared(acquires);
            }
        }

公平模式与非公平模式的主要差异就在获取许可时的机制，非公平模式直接通过自旋操作让所有线程竞争许可，从而导致了非公平。而公平模式则通过队列来实现公平机制。它们的差异就在tryAcquireShared方法，我们看公平模式的tryAcquireShared方法。实际上不同的地方就在下图中加了方框的两行代码，它会检查是否已经存在等待队列，如果已经有等待队列则返回-1，返回-1则表示让AQS同步器将当前线程进入等待队列中，队列则意味着公平。

static final class FairSync extends Sync {
        
            private static final long serialVersionUID = 2014338818796000944L;
    
            FairSync(int permits) {
        
                super(permits);
            }
            
            protected int tryAcquireShared(int acquires) {
        
                for (;;) {
        
                    // 为了保证顺序的公平，检查是否已经有等待队列了
                    if (hasQueuedPredecessors())
                        return -1;
                    int available = getState();
                    int remaining = available - acquires;
                    if (remaining < 0 ||
                        compareAndSetState(available, remaining))
                        return remaining;
                }
            }
        }

abstract static class Sync extends AbstractQueuedSynchronizer {
        
            private static final long serialVersionUID = 1192457210091910933L;
    
            Sync(int permits) {
        
                setState(permits);
            }
    
            final int getPermits() {
        
                return getState();
            }
    
            final int nonfairTryAcquireShared(int acquires) {
        
                for (;;) {
        
                    int available = getState();
                    int remaining = available - acquires;
                    if (remaining < 0 ||
                        compareAndSetState(available, remaining))
                        return remaining;
                }
            }
    
            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;
                }
            }
    
            final void reducePermits(int reductions) {
        
                for (;;) {
        
                    int current = getState();
                    int next = current - reductions;
                    if (next > current) // underflow
                        throw new Error("Permit count underflow");
                    if (compareAndSetState(current, next))
                        return;
                }
            }
    
            final int drainPermits() {
        
                for (;;) {
        
                    int current = getState();
                    if (current == 0 || compareAndSetState(current, 0))
                        return current;
                }
            }
        }

[Link 1]: https://blog.csdn.net/weixin_45970271/article/details/125267014