简介

1、具体内容：

在多线程的开发过程之中最为著名的案例就是生产者与消费者操作，该操作的主要流程如下：

  1. 生产者负责信息内容的生产;
  2. 每当生产者生产完成一项完整的信息之后消费者要从这里面取走信息;
  3. 如果生产者没有生产者则消费者要等待它生产完成，如果消费者还没有对信息进行消费，则生产者应该等待消费处理完成后再继续进行生产。

2、程序的基本实现

可以将生产者与消费者定义为两个独立的线程类对象，但是对于现在生产的数据，可以使用如下的组成：

  • 数据一: title = 王建、content = 宇宙大帅哥;
  • 数据二: title = 小高、content = 猥琐第一人;

既然生产者与消费者是两个独立的线程，那么这两个独立的线程之间就需要有一个数据的保存集中点，那么可以单独定义一个 Message 类实现数据的保存。

生产者与消费者

生产者与消费者基本程序模型 - 图1

范例: 程序基本结构

  1. package com.hanliukui;
  3. class Message{
  5.     private String title;
  6.     private String content;
  8.     public String getTitle() {
  9.         return title;
  10.     }
  12.     public void setTitle(String title) {
  13.         this.title = title;
  14.     }
  16.     public String getContent() {
  17.         return content;
  18.     }
  20.     public void setContent(String content) {
  21.         this.content = content;
  22.     }
  23. }
  26. class Producer implements Runnable{
  27.     private Message message;
  29.     public Producer(Message message){
  30.         this.message = message;
  31.     }
  33.     @Override
  34.     public void run() {
  35.         for (int i = 0; i < 100; i++) {
  37.             try {
  38.                 if (i % 2==0) {
  39.                     message.setTitle("王建");
  41.                     Thread.sleep(1000);
  42.                     message.setContent("宇宙大帅哥");
  43.                 }else {
  44.                     message.setTitle("小高");
  46.                     Thread.sleep(1000);
  47.                     message.setContent(" 猥琐第一人");
  48.                 }
  49.             } catch (InterruptedException e) {
  50.                 e.printStackTrace();
  51.             }
  52.         }
  53.     }
  54. }
  56. class Consumer implements Runnable{
  58.     private Message message;
  60.     public Consumer(Message message){
  61.         this.message = message;
  62.     }
  64.     @Override
  65.     public void run() {
  66.         for (int i = 0; i < 100; i++) {
  67.             try {
  68.                 String title = message.getTitle();
  69.                 Thread.sleep(1000);
  70.                 String content = message.getContent();
  71.                 System.out.println("消费："+title+"--"+content);
  72.             } catch (InterruptedException e) {
  73.                 e.printStackTrace();
  74.             }
  76.         }
  77.     }
  79. }
  81. public class ThreadDemo01 {
  83.     public static void main(String[] args) {
  84.         Message message = new Message();
  85.         new Thread(new Producer(message)).start();
  86.         new Thread(new Consumer(message)).start();
  87.     }
  89. }

image.png
通过整个代码的执行你会发现此时有两个主要问题：

  • 问题一：数据不同步了;
  • 问题二： 生产一个取走一个，但是发现有了重复生产和重复取出问题。

解决生产者 - 消费者同步问题

解决数据同步

如果要解决问题，首先解决的就是数据同步的处理问题，如果要想解决数据同步最简单的做法是使用 synchronized 关键字定义同步代码块或同步方法，于是这个时候对于同步的处理就可以直接在 Message 类中完成。

范例: 解决同步操作

  1. package com.hanliukui;
  3. class Message{
  5.     private String title;
  6.     private String content;
  8.     public synchronized void set(String title,String content){
  9.         this.title=title;
  10.         this.content=content;
  11.     }
  13.     public synchronized String get() {
  14.         return title+"--"+content;
  15.     }
  17. }
  20. class Producer implements Runnable{
  21.     private Message message;
  23.     public Producer(Message message){
  24.         this.message = message;
  25.     }
  27.     @Override
  28.     public void run() {
  29.         for (int i = 0; i < 100; i++) {
  30.             try {
  31.                 if (i % 2==0) {
  32.                     message.set("王建","宇宙大帅哥");
  33.                     Thread.sleep(1000);
  34.                 }else {
  35.                     Thread.sleep(1000);
  36.                     message.set("小高","猥琐第一人");
  37.                 }
  38.             } catch (InterruptedException e) {
  39.                 e.printStackTrace();
  40.             }
  41.         }
  42.     }
  43. }
  45. class Consumer implements Runnable{
  47.     private Message message;
  49.     public Consumer(Message message){
  50.         this.message = message;
  51.     }
  53.     @Override
  54.     public void run() {
  55.         for (int i = 0; i < 100; i++) {
  56.             try {
  57.                 String message = this.message.get();
  58.                 Thread.sleep(1000);
  59.                 System.out.println("消费："+message);
  60.             } catch (InterruptedException e) {
  61.                 e.printStackTrace();
  62.             }
  63.         }
  64.     }
  66. }
  68. public class ThreadDemo01 {
  70.     public static void main(String[] args) {
  71.         Message message = new Message();
  72.         new Thread(new Producer(message)).start();
  73.         new Thread(new Consumer(message)).start();
  74.     }
  76. }

image.png
在进行同步处理的时候肯定需要有一个同步的处理对象，那么此时肯定要将同步操作交由 Message 类处理是最合适的。

这个时候发现数据已经可以正常的保持一致了，但是对于重复操作的问题依然存在。

等待与唤醒机制

如果说现在要想解决生产者与消费者的问题，那么最好的解决方案就是使用等待与唤醒机制，而对于等待与唤醒的操作机制，主要依靠的是 Object 类中提供的方法处理的。

利用 Object 类解决重复操作

线程等待与唤醒

如果说现在要想解决生产者与消费者的问题，那么最好的解决方案就是使用等待与. 唤醒机制，而对于等待与唤醒的操作机制，主要依靠的是 Object 类中提供的方法处理的:

  1. // 等待
  2. public final void wait() throws InterruptedException; //死等
  3. public final void wait(long timeout) throws InterruptedException; //设置等待时间
  4. public final void wait(long timeout, int nanos) throws InterruptedException; //设置等待时间
  5. public final void notify(); // 唤醒第一个等待线程
  6. public final void notifyAll(); // 唤醒全部等待线程

如果此时有若干个等待线程的话，那么 notify()表示的是唤醒第一个等待的，而其它的线程继续等待. 而 notifyAll()表示醒所有等待的线程，哪个线程的优先级高就有可能先执行。

对于当前的问题主要的解决应该通过 Message 类完成处理。

范例: 修改 Message 类：

  1. package com.hanliukui;
  3. class Message {
  5.     private String title;
  6.     private String content;
  7.     // 用来标记状态，
  8.     // true 可以消费，不能生产
  9.     // false 可以生产，不能消费
  10.     private boolean flag = false;
  12.     public synchronized void set(String title, String content) {
  13.         try {
  14.             if (flag) {
  15.                 super.wait();
  16.             } else {
  17.                 Thread.sleep(1000);
  18.                 this.title = title;
  19.                 this.content = content;
  20.                 System.out.println("生产："+title+"--"+content);
  21.                 flag = true;
  22.                 super.notify();
  23.             }
  24.         } catch (InterruptedException e) {
  25.             e.printStackTrace();
  26.         }
  27.     }
  29.     public synchronized void get() {
  30.         try {
  31.             if (flag) {
  32.                 Thread.sleep(1000);
  33.                 System.out.println("消费："+title + "--" + content);
  34.                 flag = false;
  35.                 super.notify();
  36.             } else {
  37.                 super.wait();
  38.             }
  39.         } catch (InterruptedException e) {
  40.             e.printStackTrace();
  41.         }
  42.     }
  43. }
  46. class Producer implements Runnable {
  47.     private Message message;
  49.     public Producer(Message message) {
  50.         this.message = message;
  51.     }
  53.     @Override
  54.     public void run() {
  55.         for (int i = 0; i < 100; i++) {
  56.             if (i % 2 == 0) {
  57.                 message.set("王建"+i, "宇宙大帅哥");
  58.             } else {
  59.                 message.set("小高"+i, "猥琐第一人");
  60.             }
  61.         }
  62.     }
  63. }
  65. class Consumer implements Runnable {
  67.     private Message message;
  69.     public Consumer(Message message) {
  70.         this.message = message;
  71.     }
  73.     @Override
  74.     public void run() {
  75.         for (int i = 0; i < 100; i++) {
  76.             this.message.get();
  77.         }
  78.     }
  80. }
  82. public class ThreadDemo01 {
  84.     public static void main(String[] args) {
  85.         Message message = new Message();
  86.         new Thread(new Producer(message)).start();
  87.         new Thread(new Consumer(message)).start();
  88.     }
  90. }

注：这种处理形式就是在进行多线程开发过程之中最原始的处理方案，整个的等待、同步唤醒机制都有开发者自行通过原生代码实现控制。