多線程處理 CyclicBarrier demo

場景一

有幾個(gè)同學(xué)約好一起去食堂吃飯,各自都從各自的宿舍出發(fā),然后到宿舍樓下集合士复。當(dāng)所有的人都到了宿舍樓下之后冗荸,再一起從宿舍樓下出發(fā)前往食堂吃飯魂毁。

    public static void main(String[] args) {
        Map<String, Boolean> stringMap = new HashMap<String, Boolean>();
        stringMap.put("111", Boolean.TRUE);
        stringMap.put("112", Boolean.TRUE);
        stringMap.put("113", Boolean.TRUE);
        stringMap.put("114", Boolean.TRUE);
        stringMap.put("115", Boolean.TRUE);
        Boolean aBoolean = testMoreThread(stringMap);
        if (aBoolean) {
            System.out.println("大家都到達(dá)了宿舍樓下垮斯,一起出發(fā)吧曙旭。川陆。。");
        } else {
            System.out.println("都回宿舍了??档悠?吆视?倒庵?顾彰??");
        }
    }

處理結(jié)果

114開始從宿舍出發(fā)排龄。。流译。
114到達(dá)樓下。暂吉。嫉父。
115開始從宿舍出發(fā)篮条。雨饺。向瓷。
115到達(dá)樓下哺徊。病苗。涮总。
112開始從宿舍出發(fā)。陷寝。。
112到達(dá)樓下夫否。堰酿。蛀恩。
113開始從宿舍出發(fā)。果善。弹谁。
113到達(dá)樓下供璧。。掰盘。
111開始從宿舍出發(fā)邮偎。泻云。宠纯。
111到達(dá)樓下碍讨。。馒胆。
大家都到達(dá)了宿舍樓下沿量,一起出發(fā)吧。柑晒。欧瘪。
114開始從宿舍樓下出發(fā)。匙赞。。
114到達(dá)食堂妖碉。涌庭。。
115開始從宿舍樓下出發(fā)欧宜。坐榆。。
115到達(dá)食堂冗茸。席镀。匹中。
112開始從宿舍樓下出發(fā)。豪诲。顶捷。
112到達(dá)食堂。屎篱。服赎。
111開始從宿舍樓下出發(fā)。交播。重虑。
111到達(dá)食堂。秦士。缺厉。
113開始從宿舍樓下出發(fā)。隧土。提针。
113到達(dá)食堂。次洼。关贵。

Process finished with exit code 0

場景二:

有幾個(gè)同學(xué)約好一起去食堂吃飯,各自都從各自的宿舍出發(fā)卖毁,然后到宿舍樓下集合揖曾。有個(gè)同學(xué)到達(dá)樓下后突然想點(diǎn)外賣,然后大家都回宿舍

    public static void main(String[] args) {
        Map<String, Boolean> stringMap = new HashMap<String, Boolean>();
        stringMap.put("111", Boolean.TRUE);
        stringMap.put("112", Boolean.TRUE);
        stringMap.put("113", Boolean.TRUE);
        stringMap.put("114", Boolean.TRUE);
        stringMap.put("115", Boolean.FALSE);
        Boolean aBoolean = testMoreThread(stringMap);
        if (aBoolean) {
            System.out.println("大家都到達(dá)了宿舍樓下亥啦,一起出發(fā)吧炭剪。。翔脱。");
        } else {
            System.out.println("都回宿舍了奴拦??届吁?错妖??疚沐?暂氯?");
        }
    }

處理結(jié)果

114開始從宿舍出發(fā)。亮蛔。痴施。
114到達(dá)樓下。。辣吃。
115開始從宿舍出發(fā)动遭。。神得。
115到達(dá)樓下厘惦。。循头。突然想叫外賣绵估,回宿舍。卡骂。国裳。
112開始從宿舍出發(fā)。全跨。缝左。
112到達(dá)樓下。浓若。渺杉。
113開始從宿舍出發(fā)。挪钓。是越。
113到達(dá)樓下。碌上。倚评。
111開始從宿舍出發(fā)。馏予。天梧。
111到達(dá)樓下。霞丧。呢岗。
111回宿舍。蛹尝。后豫。
113回宿舍。突那。硬贯。
114回宿舍。陨收。。
112回宿舍。务漩。拄衰。
都回宿舍了?饵骨?翘悉??居触?妖混??

Process finished with exit code 0

多線程處理實(shí)現(xiàn)

public static Boolean testMoreThread(Map<String, Boolean> stringMap) {
        CyclicBarrierDemo sCyclicBarrier = new CyclicBarrierDemo(stringMap.size()+1, new Runnable() {
            @Override
            public void run() {
            }
        });
        try {
            for (String key : stringMap.keySet()) {
                Worker worker = new Worker(key, 5000, sCyclicBarrier, stringMap.get(key));
                worker.start();
            }
            Thread.sleep(1000);
            sCyclicBarrier.await();//工人完成工作轮洋,計(jì)數(shù)器減一
        } catch (Exception e) {
            return Boolean.FALSE;
        }
        return Boolean.TRUE;
    }
    public static class Worker extends Thread {
        String workerName;
        int workTime;
        CyclicBarrierDemo sCyclicBarrier;
        Boolean workValue;

        public Worker(String workerName, int workTime, CyclicBarrierDemo sCyclicBarrier, Boolean workValue) {
            this.workerName = workerName;
            this.workTime = workTime;
            this.sCyclicBarrier = sCyclicBarrier;
            this.workValue = workValue;
        }
        
        public void run() {
            System.out.println(workerName + "開始從宿舍出發(fā)制市。。弊予。");
            try {
                if (workValue) {
                    System.out.println(workerName + "到達(dá)樓下祥楣。。汉柒。");
                    sCyclicBarrier.await();//等待別的同學(xué)
                    System.out.println(workerName + "開始從宿舍樓下出發(fā)误褪。。碾褂。");
                    System.out.println(workerName + "到達(dá)食堂兽间。。正塌。");
                } else {
                    System.out.println(workerName + "到達(dá)樓下嘀略。。传货。突然想叫外賣屎鳍,回宿舍。问裕。逮壁。");
                    sCyclicBarrier.out();
                }
            } catch (Throwable e) {
                ExceptionUtil.printStackTrace(e);
                System.out.println(workerName + "回宿舍。粮宛。窥淆。");
                sCyclicBarrier.out();
            }
        }
    }
package com.self.ddyoung.daily.plugin;

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author <a href="mailto:sanbian@pamirs.top">Sanbian</a>
 * @version 1.0
 * @since 17/2/21 下午8:05
 */
public class CyclicBarrierDemo{

    /**
     * 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 <tt>count</tt> 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 CyclicBarrierDemo.Generation generation = new CyclicBarrierDemo.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 CyclicBarrierDemo.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();
    }

    AtomicBoolean out = new AtomicBoolean(false
    );
    public void out(){
        out.getAndSet(true);
    }

    /**
     * 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 CyclicBarrierDemo.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 (;;) {
                if(out.get()){
                    Thread.currentThread().interrupt();
                }
                try {
                    if (!timed) {
//                        if(out.get()){
//                            Thread.currentThread().interrupt();
//                        }
//                        System.out.print("awaitNanos(100)");
                        trip.awaitNanos(100);
//                        Thread.sleep(100L);
                    }
                    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 <tt>CyclicBarrier</tt> 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 CyclicBarrierDemo(int parties, Runnable barrierAction) {
        if (parties <= 0) throw new IllegalArgumentException();
        this.parties = parties;
        this.count = parties;
        this.barrierCommand = barrierAction;
    }

    /**
     * Creates a new <tt>CyclicBarrier</tt> 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 CyclicBarrierDemo(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
     * <tt>await</tt> 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
     * <tt>await</tt> 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
     *         <tt>{@link #getParties()} - 1</tt> 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 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
     * <tt>await</tt> 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
     * <tt>await</tt> 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
     *         <tt>{@link #getParties()} - 1</tt> 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
     * @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 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();
        }
    }
}

最后編輯于
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
  • 序言:七十年代末,一起剝皮案震驚了整個(gè)濱河市巍杈,隨后出現(xiàn)的幾起案子忧饭,更是在濱河造成了極大的恐慌,老刑警劉巖筷畦,帶你破解...
    沈念sama閱讀 216,744評論 6 502
  • 序言:濱河連續(xù)發(fā)生了三起死亡事件词裤,死亡現(xiàn)場離奇詭異刺洒,居然都是意外死亡,警方通過查閱死者的電腦和手機(jī)吼砂,發(fā)現(xiàn)死者居然都...
    沈念sama閱讀 92,505評論 3 392
  • 文/潘曉璐 我一進(jìn)店門逆航,熙熙樓的掌柜王于貴愁眉苦臉地迎上來,“玉大人渔肩,你說我怎么就攤上這事因俐。” “怎么了周偎?”我有些...
    開封第一講書人閱讀 163,105評論 0 353
  • 文/不壞的土叔 我叫張陵抹剩,是天一觀的道長。 經(jīng)常有香客問我蓉坎,道長澳眷,這世上最難降的妖魔是什么? 我笑而不...
    開封第一講書人閱讀 58,242評論 1 292
  • 正文 為了忘掉前任袍嬉,我火速辦了婚禮境蔼,結(jié)果婚禮上,老公的妹妹穿的比我還像新娘伺通。我一直安慰自己箍土,他們只是感情好,可當(dāng)我...
    茶點(diǎn)故事閱讀 67,269評論 6 389
  • 文/花漫 我一把揭開白布罐监。 她就那樣靜靜地躺著吴藻,像睡著了一般。 火紅的嫁衣襯著肌膚如雪弓柱。 梳的紋絲不亂的頭發(fā)上沟堡,一...
    開封第一講書人閱讀 51,215評論 1 299
  • 那天,我揣著相機(jī)與錄音矢空,去河邊找鬼航罗。 笑死,一個(gè)胖子當(dāng)著我的面吹牛屁药,可吹牛的內(nèi)容都是我干的粥血。 我是一名探鬼主播,決...
    沈念sama閱讀 40,096評論 3 418
  • 文/蒼蘭香墨 我猛地睜開眼酿箭,長吁一口氣:“原來是場噩夢啊……” “哼复亏!你這毒婦竟也來了?” 一聲冷哼從身側(cè)響起缭嫡,我...
    開封第一講書人閱讀 38,939評論 0 274
  • 序言:老撾萬榮一對情侶失蹤缔御,失蹤者是張志新(化名)和其女友劉穎,沒想到半個(gè)月后妇蛀,有當(dāng)?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體耕突,經(jīng)...
    沈念sama閱讀 45,354評論 1 311
  • 正文 獨(dú)居荒郊野嶺守林人離奇死亡笤成,尸身上長有42處帶血的膿包…… 初始之章·張勛 以下內(nèi)容為張勛視角 年9月15日...
    茶點(diǎn)故事閱讀 37,573評論 2 333
  • 正文 我和宋清朗相戀三年,在試婚紗的時(shí)候發(fā)現(xiàn)自己被綠了有勾。 大學(xué)時(shí)的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片疹启。...
    茶點(diǎn)故事閱讀 39,745評論 1 348
  • 序言:一個(gè)原本活蹦亂跳的男人離奇死亡,死狀恐怖蔼卡,靈堂內(nèi)的尸體忽然破棺而出,到底是詐尸還是另有隱情挣磨,我是刑警寧澤雇逞,帶...
    沈念sama閱讀 35,448評論 5 344
  • 正文 年R本政府宣布,位于F島的核電站茁裙,受9級特大地震影響塘砸,放射性物質(zhì)發(fā)生泄漏。R本人自食惡果不足惜晤锥,卻給世界環(huán)境...
    茶點(diǎn)故事閱讀 41,048評論 3 327
  • 文/蒙蒙 一掉蔬、第九天 我趴在偏房一處隱蔽的房頂上張望。 院中可真熱鬧矾瘾,春花似錦女轿、人聲如沸。這莊子的主人今日做“春日...
    開封第一講書人閱讀 31,683評論 0 22
  • 文/蒼蘭香墨 我抬頭看了看天上的太陽。三九已至放妈,卻和暖如春北救,著一層夾襖步出監(jiān)牢的瞬間,已是汗流浹背芜抒。 一陣腳步聲響...
    開封第一講書人閱讀 32,838評論 1 269
  • 我被黑心中介騙來泰國打工珍策, 沒想到剛下飛機(jī)就差點(diǎn)兒被人妖公主榨干…… 1. 我叫王不留,地道東北人宅倒。 一個(gè)月前我還...
    沈念sama閱讀 47,776評論 2 369
  • 正文 我出身青樓攘宙,卻偏偏與公主長得像,于是被迫代替她去往敵國和親唉堪。 傳聞我的和親對象是個(gè)殘疾皇子模聋,可洞房花燭夜當(dāng)晚...
    茶點(diǎn)故事閱讀 44,652評論 2 354

推薦閱讀更多精彩內(nèi)容