java多线程之同步、等待(wait,notify,notifyAll)

     对于JDK1.5以上的版本,如果涉及到多个线程之间的同步、等待,可以使用CyclicBarrier工具类,CyclicBarrier是JDK1.5中的同步辅助类,其应用场景是:如果有一组线程需要互相等待,直到达到某个公共屏障点,则CyclicBarrier非常有用。比如赛马游戏,可以把每匹马看做是一个线程,则每一轮赛马都必须等到所有的马全部跑完上一轮才能重新开始。

以下代码是Think in java第四版中关于CyclicBarrier的一个例子:

//: concurrency/HorseRace.java
// Using CyclicBarriers.
import java.util.concurrent.*;
import java.util.*;

class Horse implements Runnable {
    private static int counter = 0;
    private final int id = counter++;
    private int strides = 0;
    private static Random rand = new Random(47);
    private static CyclicBarrier barrier;

    public Horse(CyclicBarrier b) {
        barrier = b;
    }

    public synchronized int getStrides() {
        return strides;
    }

    public void run() {
        try {
            while (!Thread.interrupted()) {
                synchronized (this) {
                    strides += rand.nextInt(3); // Produces 0, 1 or 2
                }
                barrier.await();
            }
        } catch (InterruptedException e) {
            // A legitimate way to exit
        } catch (BrokenBarrierException e) {
            // This one we want to know about
            throw new RuntimeException(e);
        }
    }

    public String toString() {
        return "Horse " + id + " ";
    }

    public String tracks() {
        StringBuilder s = new StringBuilder();
        for (int i = 0; i < getStrides(); i++)
            s.append("*");
        s.append(id);
        return s.toString();
    }
}

public class HorseRace {
    static final int FINISH_LINE = 75;
    private List<Horse> horses = new ArrayList<Horse>();
    private ExecutorService exec = Executors.newCachedThreadPool();
    private CyclicBarrier barrier;

    public HorseRace(int nHorses, final int pause) {
        barrier = new CyclicBarrier(nHorses, new Runnable() {
            public void run() {
                StringBuilder s = new StringBuilder();
                for (int i = 0; i < FINISH_LINE; i++)
                    s.append("="); // The fence on the racetrack
                System.out.println(s);
                for (Horse horse : horses)
                    System.out.println(horse.tracks());
                for (Horse horse : horses)
                    if (horse.getStrides() >= FINISH_LINE) {
                        System.out.println(horse + "won!"+horse.getStrides());
                        exec.shutdownNow();
                        return;
                    }
                try {
                    TimeUnit.MILLISECONDS.sleep(pause);
                } catch (InterruptedException e) {
                    System.out.println("barrier-action sleep interrupted");
                }
            }
        });
        for (int i = 0; i < nHorses; i++) {
            Horse horse = new Horse(barrier);
            horses.add(horse);
            exec.execute(horse);
        }
    }

    public static void main(String[] args) {
        int nHorses = 7;
        int pause = 50;
        if (args.length > 0) { // Optional argument
            int n = new Integer(args[0]);
            nHorses = n > 0 ? n : nHorses;
        }
        if (args.length > 1) { // Optional argument
            int p = new Integer(args[1]);
            pause = p > -1 ? p : pause;
        }
        new HorseRace(nHorses, pause);
    }
} /* (Execute to see output) */// :~

 

对于JDK1.5之前的版本,我们也可以模拟多个线程等待公共屏障点:

/**
* java多线程之:赛马
* 在本程序中,每个线程代表一匹马。 在每一轮中,每匹”马“跑出随机数之内(0-4之间)的距离,
* 然后wait,主线程检测到所有的”马“都wait时,打印出截止到当前这一轮,每匹马一共跑了多少,
* 如果有”马“已到达终点,程序结束,否则,主线程将唤醒所有等待的线程,继续跑下一轮。
*
* 关键字:线程的同步、等待
*/

import java.util.Random;
import java.util.concurrent.TimeUnit;

public class Main2 {
    public static void main(String[] args) throws InterruptedException {
        final int MAX_HORSE = 5;
        final int FINAL_LINE = 30; // 终点线
        MyHorse[] arr = new MyHorse[MAX_HORSE];
        Race race = new Race(MAX_HORSE);

        for (int i = 0; i < MAX_HORSE; i++) {
            MyHorse r = new MyHorse(race);
            arr[i] = r;
            new Thread(r).start();
        }

        while (!race.over) {
            if (race.alreadyRun < race.horseNum) {
                // 这一轮还有一些马没有跑完,休眠200毫秒,等待全部的马跑完这一轮
                TimeUnit.MILLISECONDS.sleep(200);
                continue;
            }
           
            StringBuffer sbuff = new StringBuffer();
            for (int i = 0; i < FINAL_LINE; i++) {
                sbuff.append("=");
            }
            //打印出分隔符
            System.out.println(sbuff.toString());
           
           
            for (int i = 0; i < arr.length; i++) {
                //打印每一匹马截止到当前这一轮跑了多少路
                System.out.println(arr[i].print());
            }
           
           
            for (int i = 0; i < arr.length; i++) {
                //如果已经有跑完全程的,则游戏结束
                //注意,因为每匹马每次跑的步长是4以内的随机数,所以最后可能存在多匹马同时赢的情况。
                if(arr[i].step>=FINAL_LINE){
                    System.out.println("horse " + arr[i].ID + " WIN! step:" + arr[i].step);
                    race.over = true;
                }
            }
           
            synchronized (race) {
                //重置计数器
                race.reset();

                //唤醒所有的马,开始下一轮跑步
                race.notifyAll();
            }

        }
    }
}
class MyHorse implements Runnable {
    static int counter = 0;
    final int ID = counter++;
    Random rand = new Random();

    Race race;
    int step = 0;

    MyHorse( Race _r) {
        race = _r;
    }

    public String print() {
        StringBuffer s = new StringBuffer();
        for (int i = 0; i < step; i++) {
            s.append("*");
        }
        s.append(ID);
        return s.toString();

    }

    @Override
    public void run() {
        while (!race.over) {
            step += rand.nextInt(4);
            //每跑完一轮,就通知race,同时当前线程wait,等待所有的马都跑完这一轮
            race.run();
        }

    }

}

class Race {
    int horseNum;
    int alreadyRun = 0;
    boolean over = false;

    Race(int _m) {
        horseNum = _m;
    }

    public synchronized void run() {
        alreadyRun++;
        try {
            wait();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    public synchronized void reset(){
        alreadyRun = 0;
    }
}

相关文章

Comments (2)

w六月 30th, 2011 at 21:39

good

w六月 30th, 2011 at 21:40

测试一下

*