Java高并发编程详解系列-线程通信

末蓝、 2022-02-05 00:04 387阅读 0赞

>   进程间的通信，又被称为是进程内部的通信，我们都知道每个进程中有多个线程在执行，多个线程要互斥的访问共享资源的时候会发送对应的等待信号或者是唤醒线程执行等信号。那么这些信号背后还有什么样的技术支持呢？

## 两种消息处理方式 ##

### 同步阻塞式消息处理 ###

假设有这样的一个功能需要完成，每个客户端需要提交一个Event到服务器，而服务器接收到客户端请求之后开辟处理线程进行客户端请求处理操作，经过处理之后将结果返回。  
如图所示(该图来源于其他书籍)  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70]  
  这样的设计存在以下的一些缺陷

1.  通过同步的方式提交了一个Event，客户端需要长时间等待服务器返回结果
2.  如果客户端的Event的请求增加的话，服务器端的处理压力将会变大
3.  每一次客户端连接都会创建线程，这种实现方式会导致线程频繁的创建和销毁，消耗资源
4.  当客户端连接增加之后需要进行不同的业务操作，就会使得CPU的处理能力降低

### 异步非阻塞消息处理 ###

对于同步阻塞的方式来说有以上四点的缺陷，基于者四点缺陷提出了异步非阻塞的方式。每个客户端提交Event之后会立即返回。但是真实的Event会被放置在一个事件处理的队列中，在队列之后有很多的线程进行处理，这些处理都是异步进行的。最后会将结果保存到另外的集合中，客户端需要处理的结果需要再次发送请求进行二次查询。  
  两者比较而言异步非阻塞来说比同步阻塞执行的效率更高。客户端不用等待处理完成之后才会返回，提高了吞吐量和并发访问量，在这里我们还可以在服务端使用线程池的方式实现线程的重复利用，让线程的创建和使用更加高效。  
如图所示(该图来源于其他书籍)  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 1]

## 单个线程间通信 ##

从上面例子中可以看到，服务端有很多的线程从队列中获取对应的Event事件进行异步处理，这就有个问题了？这些线程是怎么从队列中获取的数据进行处理的，如何知道这个队列中此时此刻是有数据的呢？最笨的办法就是一个一个找看看队列中是否有Event，还有一个办法就是让队列自己告诉处理线程这个队列中有消息，并且还可以知道有多少Event。

### EventQueue的实现 ###

首先我们要知道对于一个队列数据结构来说其实就是一个有限制的列表，这个限制就是只能从最后插入，只能从最前面读取，最大的特点就是先进先出。对于列表来说都有大小，而队列的三种状态就是

1.  队列满
2.  队列为空
3.  队列不为空且未满

public class EventQueue { 
        private final int max;
    
        public EventQueue(int max) { 
            this.max = max;
        }
        static class Event{ 
            
        }
        private final LinkedList<Event> eventQueue = new LinkedList<>();
        
        private final static int DEFAULT_MAX_EVENT = 10;
        
        public EventQueue(){ 
            this(DEFAULT_MAX_EVENT);
        }
        
        public void offer(Event event){ 
            synchronized (eventQueue){ 
                if (eventQueue.size()>=max){ 
                    try { 
                        console("队列已满");
                        eventQueue.wait();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
                console("新事件已经提交");
                eventQueue.addLast(event);
                event.notify();
            }
        }
        
        public Event take(){ 
            synchronized (eventQueue){ 
                if (eventQueue.isEmpty()){ 
                    try { 
                        console("队列为空");
                        eventQueue.wait();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
                Event event = eventQueue.removeFirst();
                this.eventQueue.notify();
                console("事件以添加");
                return  event;
            }
        }
        private void console(String message) { 
            System.out.printf("%s:%s\n",Thread.currentThread().getName(),message);
        }
    }

在上述代码中offer方法调用它会添加一个Event到队列尾部，如果这个时候队列达到最大值，这样的话再次提交内容就会被阻塞，同样的方式使用take从队列头位置获取数据，如果队列中获取的数据为空，那么就会被阻塞。可以看到两种方式都是通过wait方法进行操作的。还有一个notify方法。首先我们知道一个线程调用wait或者是sleep方法会进入到Wait状态。而唤醒这些则需要调用notify方法进行唤醒操作。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 2]  
  通过对于线程的执行过程的分析，在很短的时间内便创建完成了10条Event数据，队列达到满的装，这个时候就要使用wait的方法进入到阻塞状态，消费者需要获取处理数据，就会通知生产者可以提交数据。

#### wait方法和notify方法 ####

在之前的时候我们分析过Object的源码，会发现这两个方法并不是线程独有的，而是通过Object对象继承重写而来。所以说这两个方式肯定是由什么特殊作用！由于是Object的方法所以说对于任何的一个对象都有对应的操作。

按照上面代码中的逻辑，当队列处于满的状态的时候会调用wait方法。对于事件队列来说是多线程共享的资源，所以进行了synchronized处理。而队列满这个状态需要调用wait方法让当前的处理event的线程进入到等待集合中，并且释放monitor的锁。

如果没有达到最大值，则需要进行队列尾部的插入操作，就需要唤醒线程，这个时候就需要使用notify方法。这个方法的主要作用就是唤醒正在执行该对象的wait方法的线程。

**注意**

1.  wait方法是可中断的方法，也就是说线程调用wait方法进入阻塞状态，可以使用interrupt方法进行打断，这个时候就会收到一个interruptException的异常，interrupt的标识也会白擦除。
2.  线程中执行了对象的wait方法之后，将这个对象加入到wait set中，每个对象的monitor都有一个关联的等待集合。
3.  线程进入到wait set之后可以使用notify方法进行唤醒操作。
4.  必须在同步方法中使用wait和notify，这两个方法执行的前提条件就是必须持有同步方法的monitor所有权，不然就会抛出一个IllegalMonitorStateException异常。例如

public void offer(Event event){ 
            synchronized (eventQueue){ 
                if (eventQueue.size()>=max){ 
                    try { 
                        console("队列已满");
                        eventQueue.wait();
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
                console("新事件已经提交");
                eventQueue.addLast(event);
                event.notify();
            }
        }

1.  同步代码块monitor必须与执行wait和notify对象是一致的，也就是说对于那个对象进行了monitor同步就需要对那个对象进行wait和notify操作

#### wait方法和sleep方法 ####

对于这两个方法来说，调用之后都会使得线程进入到阻塞状态，但是我们知道wait方法是对象Object中带有的，而Sleep则是线程独有的。两个方法都可以被中断，并且都可以收到中断异常信息。wait方法需要在同步方法中执行，而Sleep可以在任意时刻执行。在Sleep方法执行的时候线程不会释放monitor的锁，但是wait方法会释放monitor锁操作。还有最为关键的一点，就是sleep方法会自动退出阻塞，但是wait方法必须调用notify方法才退出。

## 多个线程进行通信 ##

### 消费者生产者 ###

#### notifyAll() ####

多个线程之间通信需要调用的是Object对象的另外的一个方法notifyAll()方法，这个方法与notify类似，都是唤醒wait的线程阻塞，notify唤醒的是一个线程，而notifyAll唤醒的是所有线程。

#### 问题分析 ####

在之前的EventQueue代码中，在多个线程并发操作下回出现数据不一致的问题，例如在EventClient中增加线程数

public class EventClient { 
        public static void main(String[] args) { 
            final EventQueue eventQueue = new EventQueue();
            new Thread(()->{ 
                while (true){ 
                    eventQueue.offer(new EventQueue.Event());
                }
            },"生产者").start();
            new Thread(()->{ 
                while (true){ 
                    eventQueue.offer(new EventQueue.Event());
                }
            },"生产者").start();
            new Thread(()->{ 
                while (true){ 
                    eventQueue.offer(new EventQueue.Event());
                }
            },"生产者").start();
    
            new Thread(()->{ 
                while (true){ 
                    eventQueue.take();
                    try { 
                        TimeUnit.SECONDS.sleep(10);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            },"消费者").start();
        }
    }

**出现数据不一致的情况**

1.  如果两个线程A和B同时访问take方法的时候进入到了wait方法阻塞中，线程C此时执行offer方法执行addLast方法之后唤醒A线程，A线程消费了一个Event之后唤醒了B线程，这个时候就会导致B线程执行一个空的LinkedList的removeFrist方法。也就是说在没有数据的时候还调用了removeFrist方法。
2.  在某个运行时间内队列中有5个数据，有线程A和线程B两个线程在执行offer方法时候调用了wait进入阻塞，在线程C中正在执行take方法消费Event之后唤醒了其中一个线程A的offer操作。当A线程执行完成之后addLast为5，此时又唤醒了B线程的offer方法，这个时候已经绕开了对于大小的检查，就会出现addLast的操作，此时EventQueue中的元素就会超过5个。

既然出现了上述的两种情况就需要对着两种情况进行改进操作，只需要将上面的代码中的判断改为while循环，将notify方法换成notifyAll方法即可。这样的话每次重新唤醒的时候都是所有线程公共争抢资源，不会出现几个线程独立获取资源的状况，也就不会出现数据不一致的问题。

#### Wait Set ####

在之前提到了调用了Wait方法之后线程会进入到一个wait set中，这个wait set是否是一个集合，但是我们知道集合的特性之一就是set中元素不会重复，这里我们从这个角度上出发分析一下，如果一个线程A调用了wait方法之后进入到了wait set中，这个时候又有一个线程A调用了wait方法进入到了wait set，这样的操作是不可能的，首先我们知道线程是不能重复启动的，第二点线程A是不能重复调用wait方法的。通过这两点可以分析出来，Wait Set内部实现可能是以Set作为数据结构来存储，当然每个JVM可能会作出不一样的调整。  
  这个Wait Set是被Monitor关联的，需要等到另外的线程调用了该monitor的notify方法之后才会从wait set中出来。  
**调用notify方法示例图**  
![在这里插入图片描述][20190503230049579.png]  
**调用notifyAll方法示例图**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 3]

## 总结 ##

对于消息的处理来说有同步和异步，阻塞与非阻塞之分，通过对两种消息处理机制的了解，深入理解了线程之间的通信原理。通过一个小例子了解了wait方法、notify方法、notifyAll方法。等等。了解了Wait Set机制，简单的分析了wait set内部原理。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70]: /images/20220205/f91adbc4578342afaac6cb9a48dc6521.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 1]: /images/20220205/bf0b45a478cb469a8cbb80433d5ade82.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 2]: /images/20220205/3a8bf61849ed4c1d87e42702963fa9a7.png
[20190503230049579.png]: /images/20220205/064f834881644425b415998e7d0ed6e9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25paHVpMTIz_size_16_color_FFFFFF_t_70 3]: /images/20220205/81f501b1bc4741b699e815fa16a37c5f.png