线程之间的协作

红太狼 2022-06-03 00:30 130阅读 0赞

当我们使用线程来同时运行多个任务时，可以通过使用锁(互斥)来同步两个任务的行为，从而使得一个任务不会干涉另一个任务的资源。也就是说如果两个任务交替着进入某项共享资源，你可以使用互斥来使得此时此刻只有一个任务访问这项资源。有的时候我们并不是要使线程之间互斥，而是希望它们之间能够彼此协作，使得多个任务之间可以一起工作去解决问题。

当任务协作时，关键的问题就是这些任务之间的握手。为了实现这种握手，必须要利用线程之间的基础特性：互斥。在这种情况下，互斥可以确保只有一个任务响应某个信号，这样就可以解决它们之间的竞争关系。

然而如果只是互斥并不能完成任务之间的协作，在互斥的基础上添加了一种新的途径，使得它们可以使自身挂起，直至某些外部条件发生变化，表示这个任务是时候继续执行了。这种握手可以通过Object 的方法wait() 、notify() 和notifyAll() 来安全的实现。在Java SE5 并发库中还提供了await() 和 signal() 的Condition 对象。

### wait() 与notifyAll() ###

wait() 会在等待外部条件改变时将任务挂起，并且只有在notify() 与notifyAll() 发生时，这个任务就会被唤醒去检查所发生的变化。因此wait() 提供了一种在任务之间对活动进行同步的方式。

这里需要注意的是：当我们调用sleep() 的时候锁并没有被释放，调用yield() 方法时也是这种情况，这一点很重要，当一个任务中调用了wait() 的时候，线程的执行将会被挂起，该对象上的锁将被释放，也就是说wait() 操作会释放锁。这意味着其他的任务将会获得这个锁，去执行它的任务。

wait() 有两种方式，第一种接收毫秒数作为参数，含义与sleep() 方法类似，都是指在制定的时间内“暂停执行”。但是与sleep() 不同的是对于wait() 而言：

> \[1\]:在执行wait() 前进锁是释放的  
> \[2\]:你可以通过notify() 与notifyAll() ,或者时间到期，可以从wait() 中恢复过来

另一种方式是wait() 不接受任何的参数。这种等待将会一直持续下去直至接收到notify() 与notifyAll() 的消息。

其中wait() 、notify() 与notifyAll() 有一个特殊的地方就是:它们并不是Thread 的一部分，这些方法属于基类Object。这么做是有道理的(一会你就可以体会到这么做的好处)，因为这些方法操作的锁是所有对象的一部分，你可以把wait() 放在任何同步控制的方法里，而不需要考虑这个类是否继承了Thread 类或者实现了 Runnable 接口。

### notify() 与notifyAll() ###

使用notify() 时，是指众多等待同一个锁的任务中只有一个被唤醒，如果你使用notify() 那么必须要保证被唤醒的是恰当的任务。另一方面为了使用notify() ,所有的任务都必须等待相同的条件，如果你有多个任务在等待多个不同的条件，那么你就不知道是否唤醒了恰当的任务。notifyAll() 会唤醒所有等待这个锁的任务

### 案例 ###

在这里用三个具体的案例去实现线程之间的相互协作，第一个案例就是我们在main 方法中实现子方法执行输出三次后主方法程执行输出三次，总共循环三次这样的步骤。在main() 方法中有两个线程执行，它们之间共同使用model 对象这把锁。

public class MainThread { 
         public static void main(String[] args) {
    
              ThreadModel model = new ThreadModel();
              new Thread(new Runnable() {
                  @Override
                  public void run() {
                       for(int i=0;i<3;i++)
                            model.subMethod(i+1);
                  }
              }){}.start();
    
              for(int i=0;i<3;i++){
                  model.mainMethod(i+1);
              }
         }
    }
    class ThreadModel{
         private boolean b = true;      //检查标志
         public synchronized void subMethod(int count){
              while (!b){       //当b 为 false 的时候子方法挂起，并释放锁，此时主方法将会获得该锁
                  try {
                       this.wait();
                  } catch (InterruptedException e) {
                       e.printStackTrace();
                  }
              }
             //b 为true 的时候子方法获得锁后开始进行输出，将标志置为false 并将主方法的任务唤醒
              for(int i=0;i<3;i++){
                  System.out.println("子线程执行第"+(i+1)+"次"+"...."+count);
              }
              b = false;
              this.notify();
         }
         public synchronized void mainMethod(int count){
             while (b){         //当b 为true 的时候主方法将挂起，释放锁，此时子方法获得该锁
                  try {
                       this.wait();
                  } catch (InterruptedException e) {
                       e.printStackTrace();
                  }
              }
              //b 为false 的时候主方法获得锁后开始进行输出，将标志置为true 并将子方法的任务唤醒
              for(int i = 0;i <3;i++){
                  System.out.println("主线程执行第"+(i+1)+"次"+"...."+count);
              }
              b = true;     
              this.notify();
         }
    }

> 输出：  
> 子线程执行第1次….1  
> 子线程执行第2次….1  
> 子线程执行第3次….1  
> 主线程执行第1次….1  
> 主线程执行第2次….1  
> 主线程执行第3次….1  
> 子线程执行第1次….2  
> 子线程执行第2次….2  
> 子线程执行第3次….2  
> 主线程执行第1次….2  
> 主线程执行第2次….2  
> 主线程执行第3次….2  
> 子线程执行第1次….3  
> 子线程执行第2次….3  
> 子线程执行第3次….3  
> 主线程执行第1次….3  
> 主线程执行第2次….3  
> 主线程执行第3次….3

下面这个例子使用线程池演示了两个过程：一个是将蜡涂到Car 上，一个是抛光它。抛光任务在涂蜡任务完成之前，是不能执行工作的，涂蜡任务在涂另一层蜡之前必须等待抛光任务完成。

import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    class Car{
        private boolean waxOn = false;
    
        public synchronized void waxed(){   //打蜡
            waxOn = true;
            notifyAll();
        }
        public synchronized void buffed() { //抛光
           waxOn = false;
           notifyAll();
        }
        public synchronized void waitForWaxing() throws InterruptedException {
            while (!waxOn)
                wait();
        }
        public synchronized void waitForBuffing() throws InterruptedException {
            while (waxOn)
                wait();
        }
    }
    
    class WaxOn implements Runnable{
        private Car car;
        public WaxOn(Car car){
            this.car = car;
        }
        @Override
        public void run() {
            try{
                while (true){
                    System.out.print("Wax On!");        //开始打蜡
                    Thread.sleep(200);
                    car.waxed();        //将标志置为true ，唤醒正在等待的WaxOff中 的任务
                    car.waitForBuffing();   //此时打蜡完成后将挂起等待抛光后被唤醒
                }
            } catch (InterruptedException e) {
                System.out.println("Exiting via interrupt");
            }
            System.out.println("Ending Wax On task");
        }
    }
    
    class WaxOff implements Runnable{
        private Car car;
        public WaxOff(Car car){
            this.car = car;
        }
        @Override
        public void run() {
            try{
                while (true){
                    car.waitForWaxing();    //如果waxOn 为false 时挂起，释放锁，下面任务不执行，当waxOn时被唤醒
                    System.out.print("Wax Off!");   //此时打蜡任务完成
                    Thread.sleep(200);
                    car.buffed();       //将waxOn 置为false 唤醒WaxOn 中的任务
                }
            } catch (InterruptedException e) {
                System.out.println("Exiting via interrupt");
            }
            System.out.println("Ending Wax Off task");
        }
    }
    
    public class WaxOMatic { 
        public static void main(String[] args) throws InterruptedException {
    
            Car car = new Car();
            ExecutorService exec = Executors.newCachedThreadPool();
            exec.execute(new WaxOff(car));
            exec.execute(new WaxOn(car));
            Thread.sleep(1000);      
            exec.shutdownNow();   //1s 之后终止这两个线程，调用exec.shutdownNow()时会调用控制线程的interrupt()
         }
    }

> 输出  
> Wax On!Wax Off!Wax On!Wax Off!Wax On!Exiting via interrupt  
> Ending Wax Off task  
> Exiting via interrupt  
> Ending Wax On task

下面这个示例是一个生产者与消费者的问题，只有在生产完成后才可以被消费，当消费完成后再执行生产任务，并且每次只生产与消费一个资源。

import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    class ReSource{ 
        private String name;
        private int count = 1;
        private boolean flag = false;
    
        public synchronized void set(String name){
            while(true){
                while(flag){   //循环判断标记，被唤醒后仍要判断标记，避免两个生产者任务挂起时产生数据错乱
                    try {
                        wait();
                    } catch (InterruptedException e) {
                        System.out.println("生产完成");
                    }
                }
                this.name = name + count;
                count++;
                System.out.println(Thread.currentThread().getName()+ "...生产者...." + this.name);
                flag = true;
                notifyAll();             //唤醒所有的线程，意味着消费者的线程也被唤醒，避免所有线程都处于等待状态
            }
        }
        public synchronized void out(){
            while(true){
                while(!flag){
                    try {
                        wait();
                    } catch (InterruptedException e) {
                        System.out.println("消费完成");
                    }
                }
                System.out.println(Thread.currentThread().getName()+"...消费者...." + this.name);
                flag = false;
                notifyAll();
            }
        }
    }
    
    class Producer implements Runnable{ 
        ReSource rs;
        Producer(ReSource rs){
            this.rs = rs;
        }
    
        public void run(){
            rs.set("烤鸭");
        }
    }
    
    class Consumer implements Runnable{ 
        ReSource rs;
        Consumer(ReSource rs){
            this.rs = rs;
        }
    
        public void run(){
            rs.out();
        }
    
    }
    
    public class Restaurant { 
        public static void main(String args []) throws InterruptedException {
    
            ReSource rs = new ReSource();
            Producer producer = new Producer(rs);             //多个生产者与消费者的案例
            Consumer consumer = new Consumer(rs);
            ExecutorService service = Executors.newCachedThreadPool();
    
            service.execute(new Thread(producer));
            service.execute(new Thread(producer));
            service.execute(new Thread(consumer));
            service.execute(new Thread(consumer));
            Thread.sleep(10);
            service.shutdownNow();
            System.exit(0);
        }
    }

> 输出  
> pool-1-thread-1…生产者….烤鸭1  
> pool-1-thread-4…消费者….烤鸭1  
> pool-1-thread-2…生产者….烤鸭2  
> pool-1-thread-4…消费者….烤鸭2  
> pool-1-thread-1…生产者….烤鸭3  
> pool-1-thread-4…消费者….烤鸭3  
> pool-1-thread-2…生产者….烤鸭4  
> pool-1-thread-4…消费者….烤鸭4  
> ……………………….

上面三个示例中在判断标志让线程挂起的时候我们都是使用了while() 循环，使用while() 循环的好处有很多：

>  *  你可能有多个任务出于不同的原因在等待同一个锁，而第一个唤醒任务可能会改变这种状况(即使你没有这么做，但是可能有人继承你的类去这么做)。如果出现这种情况，那么这个任务应该被再次挂起，直至其感兴趣的条件发生变化。
>  *  在这个任务从其wait() 被唤醒的时刻，有可能会在某个其他的任务中做出了改变，从而使得这个任务不能执行，或者执行其他的操作显得无关紧要。此时应该再次调用wait() 将其挂起。
>  *  也有可能某些任务出于不同的原因在等待你对象上的锁。在这种情况下，你需要检查是否已经由正确的原因唤醒，如果不是就再次挂起。

在示例三中如果不用while() 循环判断标记，当两个生产者线程都处于挂起状态时，如果一个消费者此时完成了消费任务，那么这两个生产者 都会被唤醒，如果一个生产者得到了资源并完成了生产任务，此时另一个生产者线程得到了执行权，在if() 条件下由于不对标记进行判断另一个生产者在原来挂起的基础上得到执行权它也会去执行生成的任务，但是它本来是应该被挂起的。此时就会出现问题，while() 循环则不会，在线程被唤醒后它还会循环区判断标记，这时候的标记将会引导它执行还是再度挂起。

参考书籍：  
《Java 编程思想》Bruce Eckel 著 陈昊鹏 译