Java多线程——线程之间的协作

Java多线程——线程之间的协作

摘要：本文主要学习多线程之间是如何协作的，以及如何使用wait()方法与notify()/notifyAll()方法。

部分内容来自以下博客：

https://www.cnblogs.com/hapjin/p/5492645.html

https://www.cnblogs.com/sharpxiajun/p/2295677.html

https://www.cnblogs.com/90zeng/p/java_multithread_2.html

为什么线程之间需要进行协作

在多线程并发的情况下，如果都对共享资源进行操作，那么会导致线程安全问题，所以我们使用线程的同步机制来保证多线程环境下程序的安全性，但是使用同步机制只能保证线程安全，并不能在两个线程或者多个线程之间自由切换，线程的切换完全受CPU的影响。

如果使用同步机制让两个线程交替打印1到10的数字，代码如下：

public class Demo {
    public static void main(String[] args) {
        DemoThread demoThread = new DemoThread();
        Thread a = new Thread(demoThread, "线程A");
        Thread b = new Thread(demoThread, "线程B");
        a.start();
        b.start();
    }
}

class DemoThread implements Runnable {
    private int num = 1;

    @Override
    public void run() {
        while (num <= 10) {
            synchronized (DemoThread.class) {
                if (num <= 10) {
                    System.out.println(Thread.currentThread().getName() + " >>> " + num++);
                }
            }
        }
    }
}

运行结果如下：

线程A >>> 1
线程A >>> 2
线程A >>> 3
线程A >>> 4
线程B >>> 5
线程B >>> 6
线程B >>> 7
线程B >>> 8
线程B >>> 9
线程B >>> 10

结果说明：

因为两个线程的调度完全受CPU时间片的影响，只有当一个线程运行时间结束后，另一个线程才能运行，并不能实现在线程运行的过程中进行切换。

如果我们想让两个线程交替打印数字，那么很显然同步机制是做不到的，这时候就需要两个线程的协作，让两个线程之间进行通信。

线程的等待唤醒机制

要达到上面所说的两个线程交替打印，需要两个线程进行通信，当第一个线程打印了之后，把自己锁起来，唤醒第二个线程打印，当第二个线程打印之后，也把自己锁起来，唤醒第一个线程，这样就可以实现两个线程的交替打印了。

线程的协作就是线程的通信，比如有A和B两个线程，A和B都可以独立运行，A和B有时也会做信息的交换，这就是线程的协作了。在Java里线程的协作是通过Object类里的wait()和notify()/和notifyAll()来实现的。

涉及的方法

◆　wait()

该方法会导致当前线程等待，直到其他线程调用了此线程的notify()或者notifyAll()方法。注意到wait()方法会抛出异常，所以在面我们的代码中要对异常进行捕获处理。

◆　wait(long timeout)

该方法与wait()方法类似，唯一的区别就是在指定时间内，如果没有notify或notifAll方法的唤醒，也会自动唤醒。wait(0)等效于wait()。

◆　nofity()

唤醒线程池中任意一个线程。

◆　notifyAll()

唤醒线程池中的所有线程。

方法的使用说明

这些方法都必须定义在同步中。因为这些方法是用于操作线程状态的方法，所以必须要明确到底操作的是哪个锁上的线程。

注意到上述操作线程的方法都是放在Object类中，这是因为方法都是同步锁的方法。而锁可以是任意对象，任意的对象都可以调用的方法一定定义在Object类中。

这些方法属于Object类的一部分而不是Thread类的一部分，这个咋看一下真的很奇怪，不过细细思考下，这种做法是有道理的，我们把锁机制授予对象会帮我们扩展线程应用的思路，至少把这几个方法用在任何的具有同步控制功能的方法中，而不用去考虑方法所在的类是继承了Thread还是实现了Runnable接口。
但是事实上使用这些方法还是要注意只能在同步控制方法或同步块里调用，因为这些操作都会使用到锁。

如果是在非同步的方法里调用这些方法，程序会编译通过，但是在运行时候程序会报出IllegalMonitorStateException异常，这个异常的含义是调用方法的线程在调用这些方法前必须拥有这个对象的锁，或者当前调用方法的对象锁不是之前同步时的那个锁。

使用唤醒等待实现两个线程交替执行

代码如下：

public class Demo {
    public static void main(String[] args) {
        DemoThread demoThread = new DemoThread();
        Thread a = new Thread(demoThread, "线程A");
        Thread b = new Thread(demoThread, "线程B");
        a.start();
        b.start();
    }
}

class DemoThread implements Runnable {
    private Integer num = 1;

    @Override
    public void run() {
        while (true) {
            synchronized (DemoThread.class) {
                DemoThread.class.notify();
                if (num <= 10) {
                    System.out.println(Thread.currentThread().getName() + " >>> " + num++);
                    try {
                        DemoThread.class.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }
}

运行结果如下：

线程A >>> 1
线程B >>> 2
线程A >>> 3
线程B >>> 4
线程A >>> 5
线程B >>> 6
线程A >>> 7
线程B >>> 8
线程A >>> 9
线程B >>> 10

线程的虚假唤醒

在使用wait()时，当被唤醒时有可能会被虚假唤醒，建议使用while而不是if来进行判断，即在循环中使用wait()方法。

在下面的代码中，没有在循环中使用wait()方法：

public class Demo {
    public static void main(String[] args) {
        DemoThread demoThread = new DemoThread();
        
        Thread a = new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < 4; i++) {
                    demoThread.increase();
                }
            }
        }, "线程A");
        Thread b = new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < 4; i++) {
                    demoThread.decrease();
                }
            }
        }, "线程B");
        Thread c = new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < 4; i++) {
                    demoThread.increase();
                }
            }
        }, "线程C");
        Thread d = new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < 4; i++) {
                    demoThread.decrease();
                }
            }
        }, "线程D");
        a.start();
        b.start();
        c.start();
        d.start();
    }
}

class DemoThread {
    private static Integer num = 1;
    
    public synchronized void increase() {
        if (num > 0) {
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        num++;
        System.out.print(num + " ");
        this.notifyAll();
    }
    
    public synchronized void decrease() {
        if (num == 0) {
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        num--;
        System.out.print(num + " ");
        this.notifyAll();
    }
}

运行结果如下：

0 1 0 1 2 1 0 1 2 1 0 1 2 1 2 1

可以看到即便使用了synchronized关键字，仍然出现了线程安全问题，原因如下：

在某一刻，一个负责增加的线程获得了资源，此时num为1，所以执行 this.wait(); 并等待。

下一刻，另一个负责增加的线程获得了资源，此时num仍然为1，所以再次执行 this.wait(); 并等待。

此后负责减少的线程将num减少到0并唤醒所有等待进程，两个负责增加的线程被唤醒，执行两次增加运算，导致num为2的情况产生。

解决办法就是将 if (num > 0) { 和 if (num == 0) { 中的if换成while。

wait()和sleep()方法的区别

两个方法声明的位置不同：Thread类中声明sleep() ，Object类中声明wait()。

使用方法不同：wait()可以指定时间，也可以不指定时间，sleep()必须指定时间。

调用的要求不同：sleep()可以在任何需要的场景下调用， wait()必须使用在同步代码块或同步方法中。

关于是否释放同步监视器：如果两个方法都使用在同步代码块或同步方法中，sleep()不会释放锁，wait()会释放锁。

生产者消费者问题

public class Demo {
    public static void main(String[] args) {
        Clerk clerk = new Clerk();
        Productor productor = new Productor(clerk);
        Consumer consumer = new Consumer(clerk);
        Thread productor1 = new Thread(productor, "生产者1");
        Thread productor2 = new Thread(productor, "生产者2");
        Thread consumer1 = new Thread(consumer, "消费者1");
        Thread consumer2 = new Thread(consumer, "消费者2");
        productor1.start();
        productor2.start();
        consumer1.start();
        consumer2.start();
    }
}

class Clerk {// 店员
    private int num = 0;// 产品数量

    public synchronized void product() {// 生产产品
        if (num < 2000) {
            System.out.println(Thread.currentThread().getName() + "：生产了第" + ++num + "个产品");
            notifyAll();
        } else {
            try {
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public synchronized void consume() {// 消费产品
        if (num > 0) {
            System.out.println(Thread.currentThread().getName() + ":消费了第" + num-- + "个产品");
            notifyAll();
        } else {
            try {
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

class Productor implements Runnable {// 生产者线程
    Clerk clerk;

    public Productor(Clerk clerk) {
        this.clerk = clerk;
    }

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + "开始生产产品");
        while (true) {
            clerk.product();
        }
    }
}

class Consumer implements Runnable {// 消费者线程
    Clerk clerk;

    public Consumer(Clerk clerk) {
        this.clerk = clerk;
    }

    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + "开始消费产品");
        while (true) {
            clerk.consume();
        }
    }
}