多线程之间的通信

本是古典何须时尚 2022-06-06 12:58 411阅读 0赞

## 我们通过一个简单的例子来看下多线程的通信 ##

> 我们有这样一个需求：  
> 1.建立一个公共资源Resource  
> 2.建立两个线程一个线程负责增加资源，一个线程负责取资源

class Resource{ 
        String name;
        String sex;
    
    }
    //输入
    class Input implements Runnable{ 
        Resource r;
        //一构造对象就有资源
        Input(Resource r){
            this.r = r;
        }
        @Override
        public void run() {
            int x = 0;
            while(true){
    
                if (x == 0){
                    r.name = "mike";
                    r.sex = "male";
                }
                else{
                    r.name = "莉莉";
                    r.sex = "女";
                }
                x = (x+1)%2;
            }
        }
    }
    //输出
    class Output implements Runnable{ 
        Resource r ;
        Output(Resource r){
            this.r = r;
        }
        @Override
        public void run() {
            while(true){
                System.out.println(r.name + "....." + r.sex);
            }
    
        }
    }
    public class comuThread { 
        public static void main(String[] args){
            Resource r = new Resource();
            Input in = new Input(r);
            Output out = new Output(r);
            Thread t1 = new Thread(in);
            Thread t2 = new Thread(out);
            t1.start();
            t2.start();
    
        }
    }

> 控制台输出：
> 
> 莉莉…..女  
> 莉莉…..女  
> 莉莉…..male  
> 莉莉…..male

莉莉变成男生了？，为什么会出现这种现象？

> 两个线程之间有共享数据（Resource对象），并且一个线程在操作赋值的同时，另外一个线程在获取name和sex，这就会产生安全问题，如果线程1刚赋值完name=“莉莉”值，还没赋值sex，cpu执行权被收回，线程2开始执行，这时候线程2获取的就是mike的sex值，导致上述现象发生，那我们怎么解决？“\`

### 我们想到使用同步机制将代码进行同步 ###

修改代码：

//输入
    class Input implements Runnable{ 
        Resource r;
        Object obj = new Object();
        //一构造对象就有资源
        Input(Resource r){
            this.r = r;
        }
        @Override
        public void run() {
            int x = 0;
            while(true){
                synchronized (obj){
       //加上同步锁
                    if (x == 0){
                        r.name = "mike";
                        r.sex = "male";
                    }
                    else{
                        r.name = "莉莉";
                        r.sex = "女";
                    }
                    x = (x+1)%2;
                }
    
            }
        }
    }

> 控制台输出：
> 
> 莉莉…..女  
> 莉莉…..女  
> 莉莉…..male  
> 莉莉…..male

为什么加了同步依旧出现了这样的问题？  
这样我们就需要考虑**同步的前提**：  
**就是一个锁里面是否有多个线程？即多个线程是否在一个锁里面？**  
这里我们就会发现，虽然输入数据Input在锁里面了，但是输出Output并不在同步锁里面，所以还是会引发先前的错误，即：我们只是控制了输入的同步，没有控制输出的同步。

**我们如何控制上述输出的同步？**  
我们需要找到一个锁控制两个线程，输入和输出  
**这里我们就想因为我们使用的是同一资源 r ，所以我们可以使用对象 r 作为同步锁，来控制两个线程**  
修改代码：

//输入
    class Input implements Runnable{ 
        Resource r;
        //一构造对象就有资源
        Input(Resource r){
            this.r = r;
        }
        @Override
        public void run() {
            int x = 0;
            while(true){
                synchronized (r){
       //这里加对象锁r
                    if (x == 0){
                        r.name = "mike";
                        r.sex = "male";
                    }
                    else{
                        r.name = "莉莉";
                        r.sex = "女";
                    }
                    x = (x+1)%2;
                }
    
            }
        }
    }
    //输出
    class Output implements Runnable{ 
        Resource r ;
        Output(Resource r){
            this.r = r;
        }
        @Override
        public void run() {
    
            while(true){
                synchronized (r){
       //这里加对象锁
                    System.out.println(r.name + "....." + r.sex);
                }
            }
        }
    }

控制台输出：

> 莉莉…..女  
> 莉莉…..女  
> 莉莉…..女  
> 莉莉…..女  
> 莉莉…..女  
> 莉莉…..女  
> 莉莉…..女  
> 莉莉…..女  
> 莉莉…..女  
> 莉莉…..女  
> mike…..male  
> mike…..male  
> mike…..male  
> mike…..male  
> mike…..male  
> mike…..male  
> mike…..male  
> mike…..male  
> mike…..male

这样我们就解决了上述的资源共享问题  
但是存在一个问题,我们会发现控制台怎么成片成片的输出 莉莉….女 和 mike….male

因为 **我们只是保证了读写资源的一致性，数据不会出错**  
解释：**当input拿到执行权的时候，它可能多次对Resource进行操作赋值，就导致name和sex在莉莉…女 mike….male之间切换，也就是对对象值的覆盖操作，当执行权被回收后，output取得执行权，同步后的output只能单一输出Resource中的name和sex，去的执行权的时间有长短，所以就会批量批量的输出上述控制台的结果。**

### 等待唤醒机制 ###

但是我们需要的是：**当input改变资源的时候我们就进行output的输出，也就是我们在保证数据一致性的前提下再保证线程的同步** 即：*input—->Resource（发生改变）—>output 构成一个输入输出整体*  
看具体示例图：  
![这里写图片描述][SouthEast]

**Output操作之后再将true改为false，然后进入等待状态，这时唤醒Input又可以进行输入，输入之后再进行唤醒Output线程，这样就完成了输入一次输出一次的过程。**  
看具体代码：

/** * Created by Cronous on 2017/11/9. * 线程中通讯 * 多个线程处理同一资源，但是任务却不同 * 等待唤醒机制 * wait() 让线程处于冻结状态，被wait的线程会被存储到线程池中 * notify() 唤醒线程池中的一个线程（任意的） * notifyAll()唤醒线程池中的所有线程（临时状态或这阻塞状态，具备执行资格） * 这些方法必须定义在同步中，这些方法用于操作线程状态的方法 * 必须明确到底是操作哪个类上的线程,即明确多个线程是否从属同一锁 */
    
    class Resource{ 
        String name;
        String sex;
        boolean flag = false;
    
    }
    //输入
    class Input implements Runnable{ 
        Resource r;
    
        //一构造对象就有资源
        Input(Resource r){
            this.r = r;
        }
        @Override
        public void run() {
            int x = 0;
            while(true){
                synchronized (r){
                    if(r.flag){
                        try {
                            r.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    if (x == 0){
                        r.name = "mike";
                        r.sex = "male";
                    }
                    else{
                        r.name = "莉莉";
                        r.sex = "女";
                    }
                    r.flag = true;
                    r.notify();
                }
                x = (x + 1)%2;
            }
        }
    }
    //输出
    class Output implements Runnable{ 
        Resource r ;
        Output(Resource r){
            this.r = r;
        }
        @Override
        public void run() {
            while(true){
                synchronized (r){
                    if(!r.flag){
                        try {
                            r.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    System.out.println(r.name + "....." + r.sex);
                    r.flag = false;
                    r.notify();
                }
    
            }
        }
    }
    public class comuThread { 
        public static void main(String[] args){
            Resource r = new Resource();
            Input in = new Input(r);
            Output out = new Output(r);
            Thread t1 = new Thread(in);
            Thread t2 = new Thread(out);
            t1.start();
            t2.start();
    
        }
    }

> 控制台输出：  
> 莉莉…..女  
> mike…..male  
> 莉莉…..女  
> mike…..male  
> 莉莉…..女  
> mike…..male  
> 莉莉…..女

我们的目的就达到了。

上面的代码虽然看的清楚但是很冗余，我们来优化一下代码：

class Resource{ 
        private String name;
        private String sex;
        private boolean flag = false;
        public synchronized void set(String name,String sex) {
            if(flag)
            {
       try {
       this.wait();} catch (InterruptedException e) {e.printStackTrace();}}
            this.name = name;
            this.sex = sex;
            flag = true;
            notify();
        }
        public synchronized void out(){
            if(!flag)
            {
       try {
       this.wait();} catch (InterruptedException e) {e.printStackTrace();}}
            System.out.println(this.name + "....." + this.sex);
            flag = false;
            notify();
        }
    }
    //输入
    class Input implements Runnable{ 
        Resource r;
    
        //一构造对象就有资源
        Input(Resource r){
            this.r = r;
        }
        @Override
        public void run() {
            int x = 0;
            while(true){
                if (x == 0){
                    r.set("mike","male");
                }
                else{
                    r.set("莉莉","女");
                }
                x = (x + 1)%2;
            }
        }
    }
    //输出
    class Output implements Runnable{ 
        Resource r ;
        Output(Resource r){
            this.r = r;
        }
        @Override
        public void run() {
            while(true){
                r.out();
            }
        }
    }
    public class comuThread { 
        public static void main(String[] args){
            Resource r = new Resource();
            Input in = new Input(r);
            Output out = new Output(r);
            Thread t1 = new Thread(in);
            Thread t2 = new Thread(out);
            t1.start();
            t2.start();
    
        }
    }

实际生产如上述代码，给对象提供set和get方法并对外提供，同步set和get函数  
上面的只有一个生产者和一个消费者，如果出现多个生产者多个消费者，又如何解决多线程之间的通信问题？

### 多生产者多消费者问题 ###

以上面的输入输出为原型，这里我们换个概念而已，输入为生产者输出为消费者  
我们这里使用两个生产者两个消费者，我们看程序执行结果会不会出错  
示例代码如下：

/** * Created by Cronous on 2017/11/9. * 多生产者多消费者的问题 */
    class Resource01{ 
        private String name;
        private int count = 1;
        private boolean flag = false;
        public synchronized void set(String name){
            if(flag){
       try {wait();} catch (InterruptedException e) {e.printStackTrace();}}
            this.name = name + count;
            count ++;
            System.out.println(Thread.currentThread().getName() + "..生产者.." + this.name);
            flag = true;
            notify();
        }
        public synchronized void out(){
            if(!flag){
       try {wait();} catch (InterruptedException e) {e.printStackTrace();}}
            System.out.println(Thread.currentThread().getName() + "..消费者.........." + this.name);
            flag = false;
            notify();
        }
    }
    
    class Producer implements Runnable{ 
        private Resource01 r;
        Producer(Resource01 r){
            this.r = r;
        }
        @Override
        public void run() {
            while(true){
                r.set("烤鸭");
            }
        }
    }
    class Consumer implements Runnable{ 
        private Resource01 r;
        Consumer(Resource01 r){
            this.r = r;
        }
        @Override
        public void run() {
            while(true){
                r.out();
            }
        }
    }
    
    public class ProducerConsumer { 
        public static  void main(String[] args){
            Resource01 r = new Resource01();
            Producer p = new Producer(r);
            Consumer c = new Consumer(r);
            Thread t0 = new Thread(p);
            Thread t1 = new Thread(p);
            Thread t2 = new Thread(c);
            Thread t3 = new Thread(c);
            t0.start();
            t1.start();
            t2.start();
            t3.start();
        }
    }

t0,t1为两个生产者，t2，t3为两个消费者

> 控制台输出：  
> Thread-3..消费者……….烤鸭49527  
> Thread-2..消费者……….烤鸭49527  
> Thread-1..生产者..烤鸭49528  
> Thread-3..消费者……….烤鸭49528  
> Thread-2..消费者……….烤鸭49528  
> Thread-1..生产者..烤鸭49529  
> Thread-3..消费者……….烤鸭49529  
> Thread-2..消费者……….烤鸭49529  
> Thread-3..消费者……….烤鸭49529  
> Thread-2..消费者……….烤鸭49529  
> ………………….  
> Thread-0..生产者..烤鸭41975  
> Thread-1..生产者..烤鸭41976  
> Thread-2..消费者……….烤鸭41976  
> Thread-0..生产者..烤鸭41977  
> Thread-1..生产者..烤鸭41978  
> Thread-2..消费者……….烤鸭41978  
> Thread-0..生产者..烤鸭41979  
> Thread-1..生产者..烤鸭41980

我们会发现输出居然有消费者消费同一产品多次的情况，而且还有生产者生产产品并没有被消费的情况出现  
解释：**首先我们要再次明确线程池的概念，wait(),sleep()都可以使线程处于冻结状态存放于线程池当中，notify()函数只是随机唤醒线程池当中的一个线程，因为是随机的，如果唤醒的是消费者就没有问题，但是如果又唤醒了生产者就会出现多次生产的问题，同理如果消费者消费完成之后，还是唤醒消费者那么就会出现多次消费问题。**

读懂一下代码：  
\-**1. 假设生产者 t0 得到执行权，先判断 flag=false 不需要等待，直接生产“烤鸭1”count =2 flag=true**

\-**2. 如果还是生产者 t0 获取执行权，这时候 flag=true t0 进入等待状态进入线程池； t1，t2，t3处于临时阻塞状态**

\-**3. 在2的前提下我们假设 t1这时 获取了执行权限，flag=true t1 进入等待状态进入线程池 ；t2，t3处于阻塞状态**

\-**4. 在3的前提下 t2 假设获得执行权限，flag=true 消费了 “烤鸭1”，flag=false notify()开始唤醒线程池中线程，这里我们会发现线程池中有线程 t0 t1 ，假设这里唤醒了 t0，（唤醒不代表拥有执行权限，这时的执行权还在 t2） 这时 t2又开始执行， ！flag = true t2 进入等待状态进入线程池，到此活动的线程为 t0 t3**

\-**5. 在4的前提下，假设这时候 t3 获得执行权 ！flag=true t3进入等待状态进入线程池，到此只有 t0 是活动的**

\-**6. 在5的前提下，这时候 t0 获取了执行权，因为是if()语句，不会判断标记了，直接执行下面代码，生产了“烤鸭2”flag=true 这时执行 notify() 线程池中有三个线程，任意唤醒一个，如果唤醒消费者t2，t3就没有问题，但是如果唤醒的是 t1，现在持有执行权的还是 t0，t0在判断flag，flag=true，t0进入等待状态进入线程池，t1取得执行权直接向下执行，生产“烤鸭3” ，到此生产消费就出现了问题，已经生产了3只烤鸭，但是消费者只是消费了一只烤鸭。**

#### 问题解决 ####

上述的代码当 t1唤醒之后，没有判断标记就开始生产了，所以我们需要往回判断flag，这里我们给代码加上 while() 语句  
代码：

class Resource01{
        private String name;
        private int count = 1;
        private boolean flag = false;
        public synchronized void set(String name){
            while(flag){
       try {wait();} catch (InterruptedException e) {e.printStackTrace();}}
            this.name = name + count;
            count ++;
            System.out.println(Thread.currentThread().getName() + "..生产者.." + this.name);
            flag = true;
            notify();
        }
        public synchronized void out(){
            while(!flag){
       try {wait();} catch (InterruptedException e) {e.printStackTrace();}}
            System.out.println(Thread.currentThread().getName() + "..消费者.........." + this.name);
            flag = false;
            notify();
        }
    }

但是这样又会出现死锁情况

控制台输出：

> Thread-0..生产者..烤鸭1  
> Thread-2..消费者……….烤鸭1  
> Thread-0..生产者..烤鸭2  
> Thread-3..消费者……….烤鸭2

只输出了四行，进入死锁  
解释：  
\-**1. 假设我们现有状态 t0，t1都处于等待状态在线程池中 flag=true，t2，t3，处于活动状态，这时执行权在 t2**

\-**2. t2进行了一次正常消费 flag = false 这时唤醒线程 t0 ，t2再次执行（执行权一直在t2），！flag=true t2 这时进入等待状态进入线程池，这时 t3获取执行权！flag=true，也进入等待状态进入线程池**

\-**3. 在2的基础上 flag=false t0获取执行权，生产了一只烤鸭 flag=true，notify() 假设此时唤醒 t1，t0有执行权，在判断标记，发现flag=true 进入等待状态进入线程池，t1获取执行权，判断标记，发现flag=true 进入等待进入线程池，至此四个线程全部进入等待状态，形成死锁**

*我们希望每次生产者生产结束后，唤醒一个消费者，而不是随机唤醒线程池中的任意一个*

我们发现没有唤醒对方才会导致死锁，唤醒本方没关系，有标记进行标记判断，所以我们可以唤醒所有  
使用**notifyAll()**

修改代码：

class Resource01{
        private String name;
        private int count = 1;
        private boolean flag = false;
        public synchronized void set(String name){
            while(flag){
       try {wait();} catch (InterruptedException e) {e.printStackTrace();}}
            this.name = name + count;
            count ++;
            System.out.println(Thread.currentThread().getName() + "..生产者.." + this.name);
            flag = true;
            notifyAll();
        }
        public synchronized void out(){
            while(!flag){
       try {wait();} catch (InterruptedException e) {e.printStackTrace();}}
            System.out.println(Thread.currentThread().getName() + "..消费者.........." + this.name);
            flag = false;
            notifyAll();
        }
    }

以上问题解决。  
**总结**  
1. while()判断标记，解决了线程获取执行权后，是否要运行  
2. notify()只能唤醒一个线程，如果本方唤醒了本方，没有意义，而且while判断标记+notify()会发生死锁  
3. notifyAll()解决了，本方线程一定唤醒对方线程

*这里我们会发现我们每次唤醒全部的线程，程序的逻辑合理性有待提高，我们只需要唤醒对方线程即可，如何优化上述方案？**这里jdk 1.5给我们提供了解决方案(java.util.concurrent.locks )。将同步和锁封装成对象，并将操作锁的隐式方式定义到该对象中，将隐式动作变成了显示动作***  
下面给出简单的对比前后伪代码：

//同步函数，同步代码块对锁的操作是隐式的
    Object obj = new Object();
    void show(){
        synchronized(obj){
            code.... wait() nootify() notifyAll()
        }
    }
    //lock对象显示方法
    Lock lock = new ReetrantLock();
    void show(){
        try{
            lock.lock();//获取锁
            code....// throw Exception();如果抛出异常，所以释放锁写在finally中
        }catch(e){
            e.printStackTrace();
        }finally{
            lock.unlock();
        }
    }
    //lock中监视器方法notify() notifyAll() wait()都封装在 condition对象中
    //我们看下condition接口
    interface Condition{
        await();
        signal();
        signalAll();
    }
    Lock lock = new ReetrantLock();
    Condition c1 = lock.newCondition();
    Condition c2 = lock.newCondition();

我们来改写一下上面的烤鸭实例

class Resource01{
        private String name;
        private int count = 1;
        private boolean flag = false;
        Lock lock = new ReentrantLock();//创建一个锁对象
        Condition c = lock.newCondition();//通过已有的锁获取该锁上的监视器对象
        public synchronized void set(String name){
            lock.lock();
            try{
                while(flag){
       try {c.await();} catch (InterruptedException e) {e.printStackTrace();}}
                this.name = name + count;
                count ++;
                System.out.println(Thread.currentThread().getName() + "..生产者.." + this.name);
                flag = true;
                c.signalAll();
            }catch(Exception e){
                e.printStackTrace();
            }finally {
                lock.unlock();
            }
    
        }
        public synchronized void out(){
            lock.lock();
            try {
                while(!flag){
       try {c.await();} catch (InterruptedException e) {e.printStackTrace();}}
                System.out.println(Thread.currentThread().getName() + "..消费者.........." + this.name);
                flag = false;
                c.signalAll();
            }catch (Exception e){
                e.printStackTrace();
            }finally {
                lock.unlock();
            }
    
        }
    }

以上代码仅仅做了一个替换而已，并没有多大改动，如何使用jdk的新的特性？  
以前的情况：  
\-**1. 一个锁上面的只有一个监视器，一个锁上面有多个线程，同一监视器监视所有线程，导致生产消费线程的紊乱；**  
\-**2. 如果我们一个锁上面有两个监视器就好了，一个监视生产者，一个监视消费者，这样就不会出现线程安全问题，实际上jdk1.5也确实提供了这样的方法，上面的例子已经提到了监视器Condition**

修改后的代码：

class Resource01{
        private String name;
        private int count = 1;
        private boolean flag = false;
        Lock lock = new ReentrantLock();
        Condition producer_c = lock.newCondition();//创建生产者监视器
        Condition consumer_c = lock.newCondition();//消费者监视器
        public synchronized void set(String name){
            lock.lock();
            try{
       //生产者等待
                while(flag){
       try {producer_c.await();} catch (InterruptedException e) {e.printStackTrace();}}
                this.name = name + count;
                count ++;
                System.out.println(Thread.currentThread().getName() + "..生产者.." + this.name);
                flag = true;
                consumer_c.signal();//唤醒一个消费者
            }catch(Exception e){
                e.printStackTrace();
            }finally {
                lock.unlock();
            }
    
        }
        public synchronized void out(){
            lock.lock();
            try {
                while(!flag){
       try {consumer_c.await();} catch (InterruptedException e) {e.printStackTrace();}}
                System.out.println(Thread.currentThread().getName() + "..消费者.........." + this.name);
                flag = false;
                producer_c.signal();//唤醒一个生产者
            }catch (Exception e){
                e.printStackTrace();
            }finally {
                lock.unlock();
            }
    
        }
    }

图例：  
![这里写图片描述][SouthEast 1]  
**lock接口：**替代了同步代码块或者同步函数，将同步的隐式操作变成显式操作，更为灵活，可以一个锁加上多个监视器  
**Condition接口：**替代了object中wait() notify() notifyAll()方法，将这些监视器方法进行了单独封装变成监视器对象，可以任意进行组合 await() signal() signalAll()  
以上线程之间的通信叙述结束.

[SouthEast]: /images/20220606/17d76ce6115e41d886e50c2f3d1e4267.png
[SouthEast 1]: /images/20220606/9bb84b8971d9441cae1cbfbc67be50d4.png