Java多线程编程-（4）-线程间通信机制的介绍与使用

悠悠 2022-06-07 04:24 110阅读 0赞

[Java多线程编程-（1）-线程安全和锁Synchronized概念][Java_-_1_-_Synchronized]

[Java多线程编程-（2）-可重入锁以及Synchronized的其他基本特性][Java_-_2_-_Synchronized]

[Java多线程编程-（3）-线程本地ThreadLocal的介绍与使用][Java_-_3_-_ThreadLocal]

## 线程间通信简介 ##

我们知道线程是操作系统中独立的个体，但是这个单独的个体之间没有一种特殊的处理方式使之成为一个整体，线程之间没有任何交流和沟通的话，他就是一个个单独的个体，不足以形成一个强大的交互性较强的整体。

为了提高CPU的利用率和各线程之间相互协作，Java的一种实现线程间通信的机制是：**wait/notify线程间通信**，下边就一起学习一下这种线程间的通信机制。

## 不使用等待/通知机制实现线程间通信 ##

假如，我们不使用下边需要介绍的机制，那我们如何实现两个线程之间的通信哪，下边看一段代码，实现的是两个线程向一个List里填充数据：

MyList代码：

public class MyList {
    
        private List list = new ArrayList();
    
        public void add() {
            list.add("我是元素");
        }
    
        public int size() {
            return list.size();
        }
    
    }

线程A：

public class ThreadA extends Thread { 
    
        private MyList list;
    
        public ThreadA(MyList list) {
            super();
            this.list = list;
        }
    
        @Override
        public void run() {
            try {
                for (int i = 0; i < 10; i++) {
                    list.add();
                    System.out.println("添加了" + (i + 1) + "个元素");
                    Thread.sleep(1000);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

线程B：

public class ThreadB extends Thread { 
    
        private MyList list;
    
        public ThreadB(MyList list) {
            super();
            this.list = list;
        }
    
        @Override
        public void run() {
            try {
                while (true) {
                    if (list.size() == 5) {
                        System.out.println("==5了，线程b要退出了！");
                        throw new InterruptedException();
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

测试类Test：

public class Test {
    
        public static void main(String[] args) {
            MyList myList = new MyList();
    
            ThreadA a = new ThreadA(myList);
            a.setName("A");
            a.start();
    
            ThreadB b = new ThreadB(myList);
            b.setName("B");
            b.start();
        }
    }

执行结果：

添加了1个元素
    添加了2个元素
    添加了3个元素
    添加了4个元素
    添加了5个元素
    ==5了，线程b要退出了！
    java.lang.InterruptedException
        at text.ThreadB.run(ThreadB.java:20)
    添加了6个元素
    添加了7个元素
    添加了8个元素
    添加了9个元素
    添加了10个元素

可以看出，当List集合中的数据为5个的时候线程B退出，虽然两个线程之间实现了通信，但是代码中我们的线程B是一直执行着`while(true)` 循环的，直到长度为5才终止执行，显然这种方式是很消耗资源的。所以，就需要一种机制能避免上述的操作又能实现多个线程之间的通信，这就是接下来需要学习的**“wait/notify线程间通信”**。

## 什么是等待/通知机制 ##

道理很简单，就像我们去银行办业务，进门之后取票号，等到达的时候会广播通知我们办业务一样，这就是很实际的一个场景，我们取了票号就需要等待，等业务员轮到票号的时候就会广播通知。

## Java中等待/通知机制的实现 ##

Java中对应等待/通知的方法是wait()/notify()，这两个方法都是超类Object中的方法，如下图所示：

![这里写图片描述][20171010185522291]

之所以会是超类Object中的方法，我们可以简单的理解：上几篇文章中我们知道任何对象都可以作为锁，而wait()/notify()是由锁调用的，想到这里自然可以体会到这里设计的巧妙之处。

**一、wait方法**

（1）方法wait()的作用是使当前执行代码的线程进行等待，该方法会将该线程放入”**预执行队列**“中，并且在wait()所在的代码处停止执行，直到接到通知或被中断为止。

（2）在调用wait()之前，线程必须获得该对象级别锁，这是一个很重要的地方，很多时候我们可能会忘记这一点，即只能在同步方法或同步块中调用wait()方法。

（3）还需要注意的是wait()是释放锁的，即在执行到wait()方法之后，当前线程会释放锁，当从wait()方法返回前，线程与其他线程竞争重新获得锁。

**二、notify方法**

（1）和wait()方法一样,notify()方法也要在同步块或同步方法中调用，即在调用前，线程也必须获得该对象的对象级别锁。

（2）该方法是用来通知那些可能等待该对象的对象锁的其他线程，如果有多个线程等待，则由线程规划器**随机**挑选出其中一个呈wait状态的线程，对其发出通知notify，并使它等待获取该对象的对象锁。

（3）这里需要注意的是，执行notify方法之后，当前线程不会立即释放其拥有的该对象锁，而是执行完之后才会释放该对象锁，被通知的线程也不会立即获得对象锁，而是等待notify方法执行完之后，释放了该对象锁，才可以获得该对象锁。

（3）notifyAll()通知所有等待同一共享资源的全部线程从等待状态退出，进入可运行状态，重新竞争获得对象锁。

**三、wait()/notify()方法总结**

（1）wait()/notify()要集合synchronized关键字一起使用，因为他们都需要首先获取该对象的对象锁；

（2）wait方法是释放锁，notify方法是不释放锁的；

（3）线程的四种状态如下图：

![这里写图片描述][20171010194652880]

## wait/notify线程间通信示例代码 ##

根据上述不使用wait/notify的代码改造如下：

MyList代码：

public class MyList {
    
        private static List list = new ArrayList();
    
        public static void add() {
            list.add("我是元素");
        }
    
        public static int size() {
            return list.size();
        }
    }

线程A：

public class ThreadA extends Thread { 
    
        private Object lock;
    
        public ThreadA(Object lock) {
            super();
            this.lock = lock;
        }
    
        @Override
        public void run() {
            try {
                synchronized (lock) {
                    if (MyList.size() != 5) {
                        System.out.println("wait begin " + System.currentTimeMillis());
                        lock.wait();
                        System.out.println("wait end " + System.currentTimeMillis());
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

线程B：

public class ThreadB extends Thread { 
    
        private Object lock;
    
        public ThreadB(Object lock) {
            super();
            this.lock = lock;
        }
    
        @Override
        public void run() {
            try {
                synchronized (lock) {
                    for (int i = 0; i < 10; i++) {
                        MyList.add();
                        if (MyList.size() == 5) {
                            lock.notify();
                            System.out.println("已发出通知！");
                        }
                        System.out.println("添加了" + (i + 1) + "个元素!");
                        Thread.sleep(1000);
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

测试代码：

public class Run {
    
        public static void main(String[] args) {
            try {
                Object lock = new Object();
                ThreadA a = new ThreadA(lock);
                a.start();
                Thread.sleep(50);
                ThreadB b = new ThreadB(lock);
                b.start();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

运行结果：

wait begin 1507634541467
    添加了1个元素!
    添加了2个元素!
    添加了3个元素!
    添加了4个元素!
    已发出通知！
    添加了5个元素!
    添加了6个元素!
    添加了7个元素!
    添加了8个元素!
    添加了9个元素!
    添加了10个元素!
    wait end  1507634551563

上述实例已经实现了简单的等待通知机制，并且我们也可以看到，虽然线程B在第五个元素的时候发出通知，而线程A实现线程B执行完之后才获得对象锁，这也可以说明，wait方法是释放锁的而notify方法是不释放锁的。

## 另一个案例：使用wait/notify模拟BlockingQueue阻塞队列 ##

BlockingQueue是阻塞队列，我们需要实现的是阻塞的放入和得到数据，设计思路如下：

（1）初始化队列最大长度为5；  
（2）需要新加入的时候，判断是否长度为5，如果是5则等待插入；  
（3）需要消费元素的时候，判断是否为0，如果是0则等待消费；

实现代码如下：

public class MyQueue { 
    
        //1、需要一个承装元素的集合
        private final LinkedList<Object> list = new LinkedList<>();
        //2、需要一个计数器
        private final AtomicInteger count = new AtomicInteger(0);
        //3、需要指定上限和下限
        private final int maxSize = 5;
        private final int minSize = 0;
    
        //5、初始化锁对象
        private final Object lock = new Object();
    
        /** * put方法 */
        public void put(Object obj) {
            synchronized (lock) {
                //达到最大无法添加，进入等到
                while (count.get() == maxSize) {
                    try {
                        lock.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                list.add(obj); //加入元素
                count.getAndIncrement(); //计数器增加
                System.out.println(" 元素 " + obj + " 被添加 ");
                lock.notify(); //通知另外一个阻塞的线程方法
            }
        }
    
        /** * get方法 */
        public Object get() {
            Object temp;
            synchronized (lock) {
                //达到最小，没有元素无法消费，进入等到
                while (count.get() == minSize) {
                    try {
                        lock.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                count.getAndDecrement();
                temp = list.removeFirst();
                System.out.println(" 元素 " + temp + " 被消费 ");
                lock.notify();
            }
            return temp;
        }
    
        private int size() {
            return count.get();
        }
    
        public static void main(String[] args) throws Exception {
    
            final MyQueue myQueue = new MyQueue();
            initMyQueue(myQueue);
    
            Thread t1 = new Thread(() -> {
                myQueue.put("h");
                myQueue.put("i");
            }, "t1");
    
            Thread t2 = new Thread(() -> {
                try {
                    Thread.sleep(2000);
                    myQueue.get();
                    Thread.sleep(2000);
                    myQueue.get();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }, "t2");
    
            t1.start();
            Thread.sleep(1000);
            t2.start();
    
        }
    
        private static void initMyQueue(MyQueue myQueue) {
            myQueue.put("a");
            myQueue.put("b");
            myQueue.put("c");
            myQueue.put("d");
            myQueue.put("e");
            System.out.println("当前元素个数：" + myQueue.size());
        }
    }

执行结果：

元素 a 被添加 
     元素 b 被添加 
     元素 c 被添加 
     元素 d 被添加 
     元素 e 被添加 
    当前元素个数：5
     元素 a 被消费 
     元素 h 被添加 
     元素 b 被消费 
     元素 i 被添加

## 其他注意事项 ##

（1）wait()和notify()方法要在同步块或同步方法中调用，即在调用前，线程也必须获得该对象的对象级别锁。

（2）wait方法是释放锁，notify方法是不释放锁的；

（3）notify每次唤醒wait等待状态的线程都是随机的，且每次只唤醒一个；

（4）notifAll每次唤醒wait等待状态的线程使之重新竞争获取对象锁，优先级最高的那个线程会最先执行；

（5）当线程处于wait()状态时，调用线程对象的interrupt()方法会出现InterruptedException异常；

## 其他知识点 ##

(1)进程间的通信方式：

管道(pipe)、有名管道(named pipe)、信号量(semophore)、消息队列(message queue)、信号(signal)、共享内存(shared memory)、套接字(socket)；

(2)线程程间的通信方式：

1、锁机制  
1.1 互斥锁：提供了以排它方式阻止数据结构被并发修改的方法。  
1.2 读写锁：允许多个线程同时读共享数据，而对写操作互斥。  
1.3 条件变量：可以以原子的方式阻塞进程，直到某个特定条件为真为止。

对条件测试是在互斥锁的保护下进行的。条件变量始终与互斥锁一起使用。

2、信号量机制：包括无名线程信号量与有名线程信号量  
3、信号机制：类似于进程间的信号处理。

线程间通信的主要目的是用于线程同步，所以线程没有象进程通信中用于数据交换的通信机制。

[Java_-_1_-_Synchronized]: http://blog.csdn.net/xlgen157387/article/details/77920497
[Java_-_2_-_Synchronized]: http://blog.csdn.net/xlgen157387/article/details/78005352
[Java_-_3_-_ThreadLocal]: http://blog.csdn.net/xlgen157387/article/details/78114278
[20171010185522291]: /images/20220607/42b9f5610efc4f70913077d4eebb0241.png
[20171010194652880]: /images/20220607/d186e111c971406a8a7e18179c7e020d.png