Thinking in Java---Concurrent包下的新构件学习+赛马游戏仿真

Posted 2020-06-16 acm_lkl

Java5的java.util.concurrent包下引入了大量的用于解决并发问题的新类；相对于前面那些基础的线程同步和通信的方法，这些新类是一种更高层次上的抽象，使用起来还是比较容易的．这篇博客就来学习其中的两个新类：CountDownLatch和CyclicBarrier；并使用CyclicBarrier来模拟一个简单的赛马游戏．

一.CountDownLatch
使用CountDownLatch对象时，我们需要给其设定一个初始的计数值，然后在这个对象上调用await()的任务都会阻塞，直到这个对象的计数值减为０；其它的任务可以在完成自己的工作时调用这个对象的countDown()方法来减少这个对象的计数值。所以这个类可以用于同步一个或多个任务，强制它们等待由其它任务执行的一组操作完成；一个典型的应用场景是将一个程序分解为ｎ个互相独立的可解决任务，并创建值为n的CountDownLatch，当每个任务完成时，就会在这个对象上调用countDown()．而那些等待这个问题被解决的任务在这个对象上调用await(),使自己阻塞，直到这个对象计数值减为０；另外值得注意的一点是这个对象并不会协调ｎ个任务执行的先后顺序。下面演示这种技术的一个框架示例：

package lkl;

import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * CountDownLatch用来同步一个或多个任务，强制它们
 * 等待由其它任务执行的一组操作完成
 * */

//预先执行的任务
class TaskPortion implements Runnable{
      private static int counter = 0;
      private final int id= counter++;

      //全局多个线程共享一个Random对象，这里其实牵涉到一个并发问题
      //只是实际上Random.next()本身就是线程安全的，所以不需要我们自己进行同步
      private static Random rand = new Random(47);

      private final CountDownLatch latch;
      public TaskPortion(CountDownLatch latch){
          this.latch = latch;
      }

      public void run(){
          try{
              dowork();
              latch.countDown(); //减少计数值
          }catch(InterruptedException ex){
              System.out.println(this+" 通过中断异常退出");  
          }
      }

      //线程睡眠一段时间模拟做些工作
     public void dowork() throws InterruptedException{
         TimeUnit.MILLISECONDS.sleep(rand.nextInt(2000));
         System.out.println(this+" completed");
     }

     public String toString(){
         return String.format("%1$-3d", id);
     }
}

class WaitingTask implements Runnable{
    private static int counter = 0;
    private final CountDownLatch latch;
    private final int id = counter++;
    public WaitingTask(CountDownLatch latch){
        this.latch = latch;
    }
    public void run()  {
        try{
            latch.await();; //在latch上的计数值减为０之前，都会阻塞在这里
            System.out.println(this+" completed");
        }catch(InterruptedException ex){
            System.out.println();
        }
    }
    public String toString(){
        return String.format("WatitingTask %1$-3d", id);
    }
}

public class CountDownLatchDemo {

    static final int SIZE = 100;
    public static void main(String[] args) throws Exception{
        ExecutorService exec = Executors.newCachedThreadPool();

        //所有线程都必须共享一个latch对象
        //设定初始计数值为100,即必须等待100个任务完成以后
        //在这个对象上调用await()的任务才能执行
       CountDownLatch latch = new CountDownLatch(SIZE);

       //开10个等待线程
       for(int i=0; i<10; i++){
           exec.execute(new WaitingTask(latch));
       }

       //开100个预先执行的线程
       for(int i=0; i<SIZE; i++){
           exec.execute(new TaskPortion(latch));
       }
       System.out.println("Latched all takss");
       exec.shutdown();
    }
}

二.CyclicBarrier
CyclicBarrier适用于这种情况：我们希望创建一组任务，它们并行的执行，然后在进行下一个步骤之前等待,直至所有的任务都完成（按正常的线程调度，这是不可能实现的）。它使得所有任务都在栅栏处列队，因此可以一致的向前移动。这看起来和上面的CountDownLatch类似，但是CountDownLatch只能触发一次事件，而CyclicBarrier可以多次重用．更具体的使用可以描述如下：我们创建一个指定了初始计数值为ｎ和Runnable对象为ｒ的CyclicBarrier对象，然后将其提交给n个线程，每个线程在完成当前的任务后就会调用这个对象上的await()减少计数值并且当前线程会阻塞；这样直到最后一个线程调用了await()使得计数值减为０，然后就会调用这个Cyclibarrier对象上r的run()方法，在run()方法执行完成以后，又会重置CyclicBarrier对象的计数值然后重复上面的过程。通过这样的一个过程，就达到了使多个线程一致向前移动的效果。
下面通过使用这个类仿真一个赛马游戏：

package lkl;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 模拟赛马游戏
 * 下面的程序中使用"*"表示目前马所在的位置，使用"=="表示栅栏
 * 具体的思想是：我们使用一个线程来模拟一匹马，在这个线程中我们每次都会
 * 生成一个随机数表示本次移动这匹马跑了多少步；然后使用CyclicBarrier来同步多个线程
 * 使它们能一致的向前移动，从而产生赛马的效果。我们会在CycylicBarrier的run()方法中
 * 打印出每次移动以后的情况。
 * 可以通过调整控制台的尺寸大小到只有马，看到这个仿真的效果。
 * */

//模拟马的线程
class Horse implements Runnable{
    private static int counter = 1;
    private final int id = counter++;//马的编号
    private int strides = 0; //当前马所走的步数
    private static Random rand = new Random(47);
    private static CyclicBarrier barrier;//所有线程共享一个CyclicBarrier对象

    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); //０，１，２
                }
            //阻塞直到最后一个线程也调用了这个方法
            //然后就会先执行barrier的run()任务，然后再重置计数值解除线程阻塞，然后调用这些线程
            //这样就可以保证每次都是每次都是每个线程都调度一次，然后输出一次当前情况
            //然后再进行下一轮的调用，这样就可以模拟赛马的情形了。不然因为操作系统的线程
            //调用是不均匀的，是模拟不了的。
            barrier.await();
            }
        }
        catch(InterruptedException ex){
            System.out.println(this+ " 通过中断异常退出");
        }catch(BrokenBarrierException e){//await()引起的异常
            throw new RuntimeException();
        }
    }
    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 nHorse,final int pause){

          //构造CyclicBarrier对象时需要传入一个Runnable对象用来计数值减为０的时候
          //执行,这里使用匿名内部类的形式传入的。在这里这个Runnable对象负责打印
          //出每次所有马都移动一次以后的情况。
          barrier = new CyclicBarrier(nHorse,new Runnable(){
              public void run(){

                  StringBuilder s = new StringBuilder();
                  //表示栅栏
                  for(int i=0 ; i<FINISH_LINE; i++){
                       s.append("=");
                  }
                  System.out.println(s);
                  //打印每匹马当前的位置(“*”+id表示)
                  for(Horse horse : horses){
                      System.out.println(horse.tracks());
                  }

                  //如果有那匹马越过终点线，则打印出该匹马获胜
                  //并结束游戏(结束掉所有的赛马线程)
                  for(Horse horse: horses){
                      if(horse.getStrides()>=FINISH_LINE){
                          System.out.println(horse+" won");
                          exec.shutdownNow();
                          return;
                      }
                  }
                  try{
                      //睡眠一段时间
                      TimeUnit.MILLISECONDS.sleep(pause);
                  }catch(InterruptedException ex){
                      ex.printStackTrace();
                  }
              }
          });

          //产生nHorse匹马赛跑
         for(int i=0; i<nHorse;i++){
             Horse horse = new Horse(barrier);
             horses.add(horse);
             exec.execute(horse);
         }
      }

      public static void main(String[] args){
          //默认7匹马赛跑
          int nHorses = 7;
          int pause = 200;
          new HorseRace(nHorses,pause);
      }
}