zoukankan      html  css  js  c++  java
  • CyclicBarrier 是如何做到等待多线程到达一起执行的?

      我们有些场景,是需要使用 多线各一起执行某些操作的,比如进行并发测试,比如进行多线程数据汇总。

      自然,我们可以使用 CountDownLatch, CyclicBarrier, 以及多个 Thread.join()。 虽然最终的效果都差不多,但实际却各有千秋。我们此处主要看 CyclicBarrier .

      

      概要: CyclicBarrier 使用 n 个 permit 进行初始化,当n个线程都到达后进行放行,然后进入下一个循环周期。在放行的同时,还可以设置一个执行方法,即相当于回调操作。

    一、CyclicBarrier 具体实现

      主循环等待!

        // CyclicBarrier
        /**
         * Main barrier code, covering the various policies.
         */
        private int dowait(boolean timed, long nanos)
            throws InterruptedException, BrokenBarrierException,
                   TimeoutException {
            // 使用一个 互斥锁,保证进行排队等待的安全性
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                // 使用的一 Generation 代表一生循环周期,当周期到达后,替换此值
                final Generation g = generation;
    
                // 针对异常情况,直接抛出异常,一般是用于多线程之间通信
                if (g.broken)
                    throw new BrokenBarrierException();
    
                if (Thread.interrupted()) {
                    // breakBarrier 是针对其他线程的,而 抛出的 InterruptedException 是针对当前线程的
                    // 从而达到中断标志全局可见的效果
                    breakBarrier();
                    throw new InterruptedException();
                }
    
                // 以下逻辑为进入了等待区域, count-1, 当减到0之后,就代表需要进行放行了
                int index = --count;
                // 放行
                if (index == 0) {  // tripped
                    boolean ranAction = false;
                    try {
                        final Runnable command = barrierCommand;
                        // 如果设置了回调,则立即执行回调,在当前线程中
                        if (command != null)
                            command.run();
                        ranAction = true;
                        // 循环周期迭代,此操作后,其他所有等待线程都将被返回,进入下一轮周期
                        nextGeneration();
                        return 0;
                    } finally {
                        // 未知异常,撤销当前的等待
                        if (!ranAction)
                            breakBarrier();
                    }
                }
    
                // loop until tripped, broken, interrupted, or timed out
                for (;;) {
                    try {
                        // 一直在此处进行等待,直到被唤醒,被唤醒时,则意味着有事件发生了
                        // 等待中将会释放锁,从而让其他线程进入
                        // 此处的 await() 是一个复杂的故事,因为它要保证在 notify 时的锁竞争问题
                        if (!timed)
                            trip.await();
                        else if (nanos > 0L)
                            nanos = trip.awaitNanos(nanos);
                    } catch (InterruptedException ie) {
                        if (g == generation && ! g.broken) {
                            breakBarrier();
                            throw ie;
                        } else {
                            // We're about to finish waiting even if we had not
                            // been interrupted, so this interrupt is deemed to
                            // "belong" to subsequent execution.
                            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();
            }
        }

      可以看到,主要逻辑就是在于 生命周期的迭代操作,但是这个生命周期的标志异常的简单:

        // 只有一个标识位, broken 为 true 时,发生了异常,整体退出
        private static class Generation {
            boolean broken = false;
        }

      而到达的线程数足够之后,需要进行周期迭代,只是 Generation 更换一个变量,另外就是要起到通知所有等待线程的作用:

        // CyclicBarrier
        /**
         * Updates state on barrier trip and wakes up everyone.
         * Called only while holding lock.
         */
        private void nextGeneration() {
            // signal completion of last generation
            // 先通知等待线程,但此时当前线程仍然持有锁,所以其他线程仍然处理等待状态
            // 然后再设置下一周期,直到本线程当前同步块退出之后,其他线程才可以进行工作
            // 此处依赖于 ReentrantLock
            // 此处体现 wait/notify 的锁作用域问题
            trip.signalAll();
            // set up next generation
            count = parties;
            generation = new Generation();
        }

      而调用 入口 仅是调用 dowait() 方法而已.

        // CyclicBarrier
        public int await() throws InterruptedException, BrokenBarrierException {
            try {
                return dowait(false, 0L);
            } catch (TimeoutException toe) {
                throw new Error(toe); // cannot happen
            }
        }

      CyclicBarrier 本身的等待逻辑是简单巧妙的,使用 ReentrantLock 的目的是为了实现带超时等待的效果,否则就是一个 wait/notify 机制的实现。当然 wait/notify 的逻辑还是很关键很复杂的,后续如有必要再写一文说明。

      完整代码如下:

    public class CyclicBarrier {
        /**
         * Each use of the barrier is represented as a generation instance.
         * The generation changes whenever the barrier is tripped, or
         * is reset. There can be many generations associated with threads
         * using the barrier - due to the non-deterministic way the lock
         * may be allocated to waiting threads - but only one of these
         * can be active at a time (the one to which {@code count} applies)
         * and all the rest are either broken or tripped.
         * There need not be an active generation if there has been a break
         * but no subsequent reset.
         */
        private static class Generation {
            boolean broken = false;
        }
    
        /** The lock for guarding barrier entry */
        private final ReentrantLock lock = new ReentrantLock();
        /** Condition to wait on until tripped */
        private final Condition trip = lock.newCondition();
        /** The number of parties */
        private final int parties;
        /* The command to run when tripped */
        private final Runnable barrierCommand;
        /** The current generation */
        private Generation generation = new Generation();
    
        /**
         * Number of parties still waiting. Counts down from parties to 0
         * on each generation.  It is reset to parties on each new
         * generation or when broken.
         */
        private int count;
    
        /**
         * Updates state on barrier trip and wakes up everyone.
         * Called only while holding lock.
         */
        private void nextGeneration() {
            // signal completion of last generation
            trip.signalAll();
            // set up next generation
            count = parties;
            generation = new Generation();
        }
    
        /**
         * Sets current barrier generation as broken and wakes up everyone.
         * Called only while holding lock.
         */
        private void breakBarrier() {
            generation.broken = true;
            count = parties;
            trip.signalAll();
        }
    
        /**
         * Main barrier code, covering the various policies.
         */
        private int dowait(boolean timed, long nanos)
            throws InterruptedException, BrokenBarrierException,
                   TimeoutException {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                final Generation g = generation;
    
                if (g.broken)
                    throw new BrokenBarrierException();
    
                if (Thread.interrupted()) {
                    breakBarrier();
                    throw new InterruptedException();
                }
    
                int index = --count;
                if (index == 0) {  // tripped
                    boolean ranAction = false;
                    try {
                        final Runnable command = barrierCommand;
                        if (command != null)
                            command.run();
                        ranAction = true;
                        nextGeneration();
                        return 0;
                    } finally {
                        if (!ranAction)
                            breakBarrier();
                    }
                }
    
                // loop until tripped, broken, interrupted, or timed out
                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 {
                            // We're about to finish waiting even if we had not
                            // been interrupted, so this interrupt is deemed to
                            // "belong" to subsequent execution.
                            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();
            }
        }
    
        /**
         * Creates a new {@code CyclicBarrier} that will trip when the
         * given number of parties (threads) are waiting upon it, and which
         * will execute the given barrier action when the barrier is tripped,
         * performed by the last thread entering the barrier.
         *
         * @param parties the number of threads that must invoke {@link #await}
         *        before the barrier is tripped
         * @param barrierAction the command to execute when the barrier is
         *        tripped, or {@code null} if there is no action
         * @throws IllegalArgumentException if {@code parties} is less than 1
         */
        public CyclicBarrier(int parties, Runnable barrierAction) {
            if (parties <= 0) throw new IllegalArgumentException();
            this.parties = parties;
            this.count = parties;
            this.barrierCommand = barrierAction;
        }
    
        /**
         * Creates a new {@code CyclicBarrier} that will trip when the
         * given number of parties (threads) are waiting upon it, and
         * does not perform a predefined action when the barrier is tripped.
         *
         * @param parties the number of threads that must invoke {@link #await}
         *        before the barrier is tripped
         * @throws IllegalArgumentException if {@code parties} is less than 1
         */
        public CyclicBarrier(int parties) {
            this(parties, null);
        }
    
        /**
         * Returns the number of parties required to trip this barrier.
         *
         * @return the number of parties required to trip this barrier
         */
        public int getParties() {
            return parties;
        }
    
        /**
         * Waits until all {@linkplain #getParties parties} have invoked
         * {@code await} on this barrier.
         *
         * <p>If the current thread is not the last to arrive then it is
         * disabled for thread scheduling purposes and lies dormant until
         * one of the following things happens:
         * <ul>
         * <li>The last thread arrives; or
         * <li>Some other thread {@linkplain Thread#interrupt interrupts}
         * the current thread; or
         * <li>Some other thread {@linkplain Thread#interrupt interrupts}
         * one of the other waiting threads; or
         * <li>Some other thread times out while waiting for barrier; or
         * <li>Some other thread invokes {@link #reset} on this barrier.
         * </ul>
         *
         * <p>If the current thread:
         * <ul>
         * <li>has its interrupted status set on entry to this method; or
         * <li>is {@linkplain Thread#interrupt interrupted} while waiting
         * </ul>
         * then {@link InterruptedException} is thrown and the current thread's
         * interrupted status is cleared.
         *
         * <p>If the barrier is {@link #reset} while any thread is waiting,
         * or if the barrier {@linkplain #isBroken is broken} when
         * {@code await} is invoked, or while any thread is waiting, then
         * {@link BrokenBarrierException} is thrown.
         *
         * <p>If any thread is {@linkplain Thread#interrupt interrupted} while waiting,
         * then all other waiting threads will throw
         * {@link BrokenBarrierException} and the barrier is placed in the broken
         * state.
         *
         * <p>If the current thread is the last thread to arrive, and a
         * non-null barrier action was supplied in the constructor, then the
         * current thread runs the action before allowing the other threads to
         * continue.
         * If an exception occurs during the barrier action then that exception
         * will be propagated in the current thread and the barrier is placed in
         * the broken state.
         *
         * @return the arrival index of the current thread, where index
         *         {@code getParties() - 1} indicates the first
         *         to arrive and zero indicates the last to arrive
         * @throws InterruptedException if the current thread was interrupted
         *         while waiting
         * @throws BrokenBarrierException if <em>another</em> thread was
         *         interrupted or timed out while the current thread was
         *         waiting, or the barrier was reset, or the barrier was
         *         broken when {@code await} was called, or the barrier
         *         action (if present) failed due to an exception
         */
        public int await() throws InterruptedException, BrokenBarrierException {
            try {
                return dowait(false, 0L);
            } catch (TimeoutException toe) {
                throw new Error(toe); // cannot happen
            }
        }
    
        /**
         * Waits until all {@linkplain #getParties parties} have invoked
         * {@code await} on this barrier, or the specified waiting time elapses.
         *
         * <p>If the current thread is not the last to arrive then it is
         * disabled for thread scheduling purposes and lies dormant until
         * one of the following things happens:
         * <ul>
         * <li>The last thread arrives; or
         * <li>The specified timeout elapses; or
         * <li>Some other thread {@linkplain Thread#interrupt interrupts}
         * the current thread; or
         * <li>Some other thread {@linkplain Thread#interrupt interrupts}
         * one of the other waiting threads; or
         * <li>Some other thread times out while waiting for barrier; or
         * <li>Some other thread invokes {@link #reset} on this barrier.
         * </ul>
         *
         * <p>If the current thread:
         * <ul>
         * <li>has its interrupted status set on entry to this method; or
         * <li>is {@linkplain Thread#interrupt interrupted} while waiting
         * </ul>
         * then {@link InterruptedException} is thrown and the current thread's
         * interrupted status is cleared.
         *
         * <p>If the specified waiting time elapses then {@link TimeoutException}
         * is thrown. If the time is less than or equal to zero, the
         * method will not wait at all.
         *
         * <p>If the barrier is {@link #reset} while any thread is waiting,
         * or if the barrier {@linkplain #isBroken is broken} when
         * {@code await} is invoked, or while any thread is waiting, then
         * {@link BrokenBarrierException} is thrown.
         *
         * <p>If any thread is {@linkplain Thread#interrupt interrupted} while
         * waiting, then all other waiting threads will throw {@link
         * BrokenBarrierException} and the barrier is placed in the broken
         * state.
         *
         * <p>If the current thread is the last thread to arrive, and a
         * non-null barrier action was supplied in the constructor, then the
         * current thread runs the action before allowing the other threads to
         * continue.
         * If an exception occurs during the barrier action then that exception
         * will be propagated in the current thread and the barrier is placed in
         * the broken state.
         *
         * @param timeout the time to wait for the barrier
         * @param unit the time unit of the timeout parameter
         * @return the arrival index of the current thread, where index
         *         {@code getParties() - 1} indicates the first
         *         to arrive and zero indicates the last to arrive
         * @throws InterruptedException if the current thread was interrupted
         *         while waiting
         * @throws TimeoutException if the specified timeout elapses.
         *         In this case the barrier will be broken.
         * @throws BrokenBarrierException if <em>another</em> thread was
         *         interrupted or timed out while the current thread was
         *         waiting, or the barrier was reset, or the barrier was broken
         *         when {@code await} was called, or the barrier action (if
         *         present) failed due to an exception
         */
        public int await(long timeout, TimeUnit unit)
            throws InterruptedException,
                   BrokenBarrierException,
                   TimeoutException {
            return dowait(true, unit.toNanos(timeout));
        }
    
        /**
         * Queries if this barrier is in a broken state.
         *
         * @return {@code true} if one or more parties broke out of this
         *         barrier due to interruption or timeout since
         *         construction or the last reset, or a barrier action
         *         failed due to an exception; {@code false} otherwise.
         */
        public boolean isBroken() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return generation.broken;
            } finally {
                lock.unlock();
            }
        }
    
        /**
         * Resets the barrier to its initial state.  If any parties are
         * currently waiting at the barrier, they will return with a
         * {@link BrokenBarrierException}. Note that resets <em>after</em>
         * a breakage has occurred for other reasons can be complicated to
         * carry out; threads need to re-synchronize in some other way,
         * and choose one to perform the reset.  It may be preferable to
         * instead create a new barrier for subsequent use.
         */
        public void reset() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                breakBarrier();   // break the current generation
                nextGeneration(); // start a new generation
            } finally {
                lock.unlock();
            }
        }
    
        /**
         * Returns the number of parties currently waiting at the barrier.
         * This method is primarily useful for debugging and assertions.
         *
         * @return the number of parties currently blocked in {@link #await}
         */
        public int getNumberWaiting() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                return parties - count;
            } finally {
                lock.unlock();
            }
        }
    }
    View Code

      

    二、简单看一下 CountDownLatch 的同时等待实现

      CountDownLatch 会在初始化时,申请 n 个 permit, 调用 await() 进行阻塞, 直到 permit=0 时,await() 才进行返回。每调用一次 countDown(); permit 都会减1直到为0止;

        // CountDownLatch.await()  等待
        public void await() throws InterruptedException {
            // 仅是去尝试获取一个而已
            sync.acquireSharedInterruptibly(1);
        }
        
        // CountDownLatch.countDown() 释放锁, 当 permit=0 后,放行 await() 
        public void countDown() {
            // 此处仅是委托给了 AQS 进行释放、通知处理
            sync.releaseShared(1);
        }
        
        // CountDownLatch 内部锁实现的是否可以持有锁的逻辑
        /**
         * Synchronization control For CountDownLatch.
         * Uses AQS state to represent 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) {
                // 只要 state=0, 都可以放行
                return (getState() == 0) ? 1 : -1;
            }
    
            // 释放锁 countDown 逻辑, 做减1操作
            protected boolean tryReleaseShared(int releases) {
                // Decrement count; signal when transition to zero
                for (;;) {
                    int c = getState();
                    // 如果已经被释放,则直接返回
                    if (c == 0)
                        return false;
                    // 忽略传入值 releases, 只做减1操作, 所以 state 必定有等于0的时候
                    int nextc = c-1;
                    if (compareAndSetState(c, nextc))
                        // 只有等于0, 才能进行真正的释放通知操作
                        return nextc == 0;
                }
            }
        }

      可以看出, CountDownLatch 的同时等待实现更加简单,几乎都是依赖于 AQS 进行实现。同样,从实际效果来说,也是一个 wait/notify 的实现。只是此处的 notify 执行完之后就释放了锁,即无法保证 notify 之后的线程安全性。

      上面两个工具也都是AQS实现的,由此也可知AQS的重要性!   

    唠叨: 论 wait/notify 机制的安全性!

  • 相关阅读:
    URL中增加BASE64加密的字符串引起的问题(java.net.MalformedURLException:Illegal character in URL)
    读《暗时间》总结
    假设写一个android桌面滑动切换屏幕的控件(一)
    JDBC Connection Reset问题分析
    深度学习工具caffe具体安装指南
    TS2
    TS 函数解析
    typescript
    响应式网页设计:rem、em设置网页字体大小自适应
    一看就懂得移动端rem布局、rem如何换算
  • 原文地址:https://www.cnblogs.com/yougewe/p/11690524.html
Copyright © 2011-2022 走看看