4.3 生产者和消费者问题-Java语言基础8-Java多线程

当前位置:　首页>> 技术小册>> Java语言基础8-Java多线程

- 等待唤醒机制可以解决经典的 生产者与消费者 的问题。
- 生产者与消费者问题，也称有限缓冲问题，是一个多线程同步问题的经典案例。该问题描述了两个（多个）共享固定大小缓冲区的线程——即所谓的 生产者 和 消费者 ——在实际运行时会发生的问题。生产者的主要作用是生成一定量的数据放到缓冲区中，然后重复此过程。与此同时，消费者也在缓冲区消耗这些数据。该问题的关键就是要保证生产者不会在缓冲区满时加入数据，消费者也不会在缓冲区中空时消耗数据。

![](/uploads/images/20230725/2f5fd934cc3cee10cdbb2b285d4c4f43.png)

-  生产者与消费者问题中其实隐含了两个问题： 
  - ① 线程安全问题：因为生产者与消费者共享数据缓冲区，不过这个问题可以使用同步解决。
  - ② 线程的协调工作问题： 
    - 要解决该问题，就必须让生产者线程在缓冲区满时等待( wait )，暂停进入阻塞状态，等到下次消费者消耗了缓冲区中的数据的时候，通知( notify )正在等待的线程恢复到就绪状态，重新开始往缓冲区添加数据。
    - 同样，也可以让消费者线程在缓冲区空时进入等待( wait )，暂停进入阻塞状态，等到生产者往缓冲区添加数据之后，再通知( notify )正在等待的线程恢复到就绪状态。
    - 通过这样的通信机制来解决此类问题。

-  示例：使用同步代码块和等待唤醒机制解决生产者和消费者问题（单生产者和单消费者）

```bash
package com.github.thread.demo14;

/**
 * 资源
 * 
 * @author maxiaoke.com
 * @version 1.0
 * 
 */
public class Resource {
    // 生产商品的计数
    int count;
    // 标记，如果flag是true，表示生产好了，等待消费；如果是false，表示消费好了，等待生产
    boolean flag;
}
```

```bash
package com.github.thread.demo14;

/**
 * 生产者
 * 
 * @author maxiaoke.com
 * @version 1.0
 * 
 */
public class Producer implements Runnable {

private final Resource r;

public Producer(Resource r) {
        this.r = r;
    }

@Override
    public void run() {
        while (true) {
            synchronized (this.r) {
                if (this.r.flag) { // 如果flag为true，表示生产好了，等待消费
                    try {
                        // 修改标志位
                        this.r.flag = false;
                        // 等待
                        this.r.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } // 如果flag为false，表示消费好了，等待生产
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                this.r.count++;
                System.out.println("生产第" + this.r.count + "个");
                // 修改标志位
                this.r.flag = true;
                // 唤醒消费者线程
                this.r.notify();
            }
        }
    }
}
```

```bash
package com.github.thread.demo14;

/**
 * 消费者
 * 
 * @author maxiaoke.com
 * @version 1.0
 * 
 */
public class Consumer implements Runnable {

private final Resource r;

public Consumer(Resource r) {
        this.r = r;
    }

@Override
    public void run() {
        while (true) {
            synchronized (this.r) {
                if (this.r.flag) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("消费第" + this.r.count + "个");

try {
                        this.r.flag = false;
                        this.r.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                this.r.flag = true;
                this.r.notify();
            }
        }
    }
}
```

```bash
package com.github.thread.demo14;

/**
 * @author maxiaoke.com
 * @version 1.0
 * 
 */
public class Test {
    public static void main(String[] args) {
        Resource resource = new Resource();

Producer producer = new Producer(resource);
        Consumer consumer = new Consumer(resource);

new Thread(producer).start();
        new Thread(consumer).start();
    }
}
```

示例：使用同步方法和等待唤醒机制解决生产者和消费者问题（单生产者和单消费者）

```bash
package com.github.thread.demo15;

/**
 * 资源
 *
 * @author maxiaoke.com
 * @version 1.0
 * 
 */
public class Resource {
    // 生产商品的计数
    private int count;
    // 标记，如果flag是true，表示生产好了，等待消费；如果是false，表示消费好了，等待生产
    private boolean flag;

/**
     * 供生产者生产的方法
     */
    public synchronized void produce() {
        // flag为true，生产完成，等待消费
        if (this.flag) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        this.count++;
        System.out.println("生产第" + this.count + "个");
        // 修改标志位，生产完成
        this.flag = true;
        this.notify();
    }

/**
     * 供消费者消费的方法
     */
    public synchronized void consume() {
        // flag为false，消费完成，等待生产
        if (!this.flag) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("消费第" + this.count + "个");
        // 修改标志位，消费完成
        this.flag = false;
        this.notify();
    }
}
```

```bash
package com.github.thread.demo15;

/**
 * @author maxiaoke.com
 * @version 1.0
 * 
 */
public class Producer implements Runnable {

private final Resource resource;

public Producer(Resource resource) {
        this.resource = resource;
    }

@Override
    public void run() {
        while (true) {
            this.resource.produce();
        }
    }
}
```

```bash
package com.github.thread.demo15;

/**
 * @author maxiaoke.com
 * @version 1.0
 * 
 */
public class Consumer implements Runnable {

private final Resource resource;

public Consumer(Resource resource) {
        this.resource = resource;
    }

@Override
    public void run() {
        while (true) {
            this.resource.consume();
        }
    }
}
```

```bash
package com.github.thread.demo15;

/**
 * @author maxiaoke.com
 * @version 1.0
 * 
 */
public class Test {
    public static void main(String[] args) {
        Resource resource = new Resource();

new Thread(new Producer(resource)).start();
        new Thread(new Consumer(resource)).start();
    }
}
```

示例：使用同步方法和等待唤醒机制解决生产者和消费者问题（多生产者和多消费者）

```bash
package com.github.thread.demo16;

/**
     * 供生产者生产的方法
     */
    public synchronized void produce() {
        // flag为true，生产完成，等待消费
        while (this.flag) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        this.count++;
        System.out.println("生产第" + this.count + "个");
        // 修改标志位，生产完成
        this.flag = true;
        this.notifyAll();
    }

/**
     * 供消费者消费的方法
     */
    public synchronized void consume() {
        // flag为false，消费完成，等待生产
        while (!this.flag) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("消费第" + this.count + "个");
        // 修改标志位，消费完成
        this.flag = false;
        this.notifyAll();
    }
}
```

```bash
package com.github.thread.demo16;

/**
 * @author maxiaoke.com
 * @version 1.0
 * 
 */
public class Producer implements Runnable {

private final Resource resource;

public Producer(Resource resource) {
        this.resource = resource;
    }

@Override
    public void run() {
        while (true) {
            this.resource.produce();
        }
    }
}
```

```bash
package com.github.thread.demo16;

/**
 * @author maxiaoke.com
 * @version 1.0
 * 
 */
public class Consumer implements Runnable {

private final Resource resource;

public Consumer(Resource resource) {
        this.resource = resource;
    }

@Override
    public void run() {
        while (true) {
            this.resource.consume();
        }
    }
}
```

```bash
package com.github.thread.demo16;

/**
 * @author maxiaoke.com
 * @version 1.0
 * 
 */
public class Consumer implements Runnable {

private final Resource resource;

public Consumer(Resource resource) {
        this.resource = resource;
    }

@Override
    public void run() {
        while (true) {
            this.resource.consume();
        }
    }
}
```

该分类下的相关小册推荐：

Java语言基础12-网络编程

Java并发编程实战

Java语言基础13-类的加载和反射

Mybatis合辑3-Mybatis动态SQL

Java语言基础5-面向对象初级

Java语言基础3-流程控制

Java语言基础14-枚举和注解

深入理解Java虚拟机

Mybatis合辑2-Mybatis映射文件

Mybatis合辑5-注解、扩展、SQL构建

Java必知必会-Maven初级

Java面试指南