单线程间通信多线程间通信

朱雀 2024-03-16 11:51 89阅读 0赞

###  单线程间通信  ###

<table> 
 <thead> 
  <tr> 
   <th align="left">首先实现一个EventQueue，该Queue有如下三种状态：</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">·<code>队列满</code>——最多可容纳多少个Event，好比一个系统最多同时能够受理多少业务一样；</td> 
  </tr> 
  <tr> 
   <td align="left">·<code>队列</code>空——当所有的Event都被处理并且没有新的Event被提交的时候，此时队列将是空的状态；</td> 
  </tr> 
  <tr> 
   <td align="left">·<code>有Event但是没有</code>满——有新的Event被提交，但是此时没有到达队列的上限。</td> 
  </tr> 
 </tbody> 
</table>

public class EventQueue {
        
        private final int 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 EventQueue(int max) {
        
            this.max = max;
        }
    
        public void offer(Event event) {
        
            synchronized (eventQueue) {
        
                if (eventQueue.size() >= max) {
        
                    try {
        
                        console(" the queue is full.");
                        eventQueue.wait();
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                }
                console(" the new event is submitted");
                eventQueue.addLast(event);
                eventQueue.notify();
            }
        }
    
        public Event take() {
        
            synchronized (eventQueue) {
        
                if (eventQueue.isEmpty()) {
        
                    try {
        
                        console(" the queue is empty.");
                        eventQueue.wait();
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                }
                Event event = eventQueue.removeFirst();
                this.eventQueue.notify();
                console(" the event " + event + " is handled.");
                return event;
            }
        }
    
        private void console(String message) {
        
            System.out.printf("%s：%s\n", currentThread().getName(), message);
        }
    }

<table> 
 <thead> 
  <tr> 
   <th align="left">上述代码中，在EventQueue中定义了一个队列，offer方法会提交一个Event至队尾，如果此时队列已经满了，那么提交的线程将会被<code>阻塞</code>，这是调用了wait方法的结果。同样take方法会从队头获取数据，如果队列中没有可用数据，那么工作线程就会被阻塞，这也是调用wait方法的直接结果。此外，还可以看到一个notify方法，该方法的作用是<code>唤醒</code>那些曾经执行monitor的wait方法而进入阻塞的线程。</th> 
  </tr> 
 </thead> 
</table>

public class EventClient {
        
        public static void main(String[] args) {
        
            final EventQueue eventQueue = new EventQueue();
            new Thread(() -> {
        
                for (; ; ) {
        
                    eventQueue.offer(new EventQueue.Event());
                }
            }, "Producer").start();
    
            new Thread(() -> {
        
                for (; ; ) {
        
                    eventQueue.take();
                    try {
        
                        TimeUnit.MILLISECONDS.sleep(10);
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                }
            }, "Consumer").start();
        }
    }

<table> 
 <thead> 
  <tr> 
   <th align="left">Producer线程模拟提交Event的客户端几乎没有任何的延迟，而Consumer线程则用于模拟处理请求的工作线程（上面的EventQueue目前只支持一个线程的Producer和一个线程的Consumer，也就是<code>单线程间的通信</code>）。</th> 
  </tr> 
 </thead> 
</table>

Producer： the new event is submitted
    Producer： the new event is submitted
    Producer： the new event is submitted
    Producer： the new event is submitted
    Producer： the new event is submitted
    Producer： the new event is submitted
    Producer： the new event is submitted
    Producer： the new event is submitted
    Producer： the new event is submitted
    Producer： the new event is submitted
    Producer： the queue is full.
    Consumer： the event thread.pcpattern.single.EventQueue$Event@4e879704 is handled.
    Producer： the new event is submitted
    Producer： the queue is full.
    Consumer： the event thread.pcpattern.single.EventQueue$Event@66e5ec31 is handled.
    Producer： the new event is submitted
    Producer： the queue is full.
    Consumer： the event thread.pcpattern.single.EventQueue$Event@64056de6 is handled.
    Producer： the new event is submitted
    Producer： the queue is full.
    Consumer： the event thread.pcpattern.single.EventQueue$Event@84a769f is handled.
    Producer： the new event is submitted
    Producer： the queue is full.
    Consumer： the event thread.pcpattern.single.EventQueue$Event@7a87ef25 is handled.
    Producer： the new event is submitted
    Producer： the queue is full.
    Consumer： the event thread.pcpattern.single.EventQueue$Event@334a0020 is handled.
    Producer： the new event is submitted
    Producer： the queue is full.
    Consumer： the event thread.pcpattern.single.EventQueue$Event@7cef31f2 is handled.
    Producer： the new event is submitted
    Producer： the queue is full.
    Consumer： the event thread.pcpattern.single.EventQueue$Event@27ca5e7e is handled.
    Producer： the new event is submitted
    Producer： the queue is full.
    ...

<table> 
 <thead> 
  <tr> 
   <th align="left">通过上述的输出日志可以看出，Producer线程很快就提交了10个Event数据，此时队列已经满了，那么它将会执行eventQueue的wait方法进而进入阻塞状态，Consumer线程由于要处理数据，所以会花费大概10毫秒的时间来处理其中的一条数据，然后通知Producer线程可以继续提交数据了，如此循环往复。</th> 
  </tr> 
 </thead> 
</table>

<table> 
 <thead> 
  <tr> 
   <th align="left">讲解了两个线程间的通信，只有一个线程对EventQueue进行offer操作，也只有一个线程对EventQueue进行take操作，如果多个线程同时进行take或者offer，那么上面的程序就会出现问题。</th> 
  </tr> 
 </thead> 
</table>

###  多线程间通信  ###

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

<table> 
 <thead> 
  <tr> 
   <th align="left">只需要将如上代码的临界值的判断if更改为<code>while</code>，将notify更改为<code>notifyAll</code>即可</th> 
  </tr> 
 </thead> 
</table>

public class EventQueuePC {
        
        private final int max;
    
        static class Event {
        
        }
    
        private final LinkedList<Event> eventQueue = new LinkedList<>();
        private final static int DEFAULT_MAX_EVENT = 10;
    
        public EventQueuePC() {
        
            this(DEFAULT_MAX_EVENT);
        }
    
        public EventQueuePC(int max) {
        
            this.max = max;
        }
    
        private void console(String message) {
        
            System.out.printf("%s：%s\n", currentThread().getName(), message);
        }
    
        public void offerSafe(Event event) {
        
            synchronized (eventQueue) {
        
                while (eventQueue.size() >= max) {
        
                    try {
        
                        console(" the queue is full.");
                        eventQueue.wait();
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                }
                console(" the new event is submitted");
                eventQueue.addLast(event);
                eventQueue.notifyAll();
            }
        }
    
        public Event takeSafe() {
        
            synchronized (eventQueue) {
        
                while (eventQueue.isEmpty()) {
        
                    try {
        
                        console(" the queue is empty.");
                        eventQueue.wait();
                    } catch (InterruptedException e) {
        
                        e.printStackTrace();
                    }
                }
                Event event = eventQueue.removeFirst();
                this.eventQueue.notifyAll();
                console(" the event " + event + " is handled.");
                return event;
            }
        }
    }

\-----------------------------------------------------------------------------读书笔记摘自书名：Java高并发编程详解：多线程与架构设计 作者：汪文君