生产者和消费者问题学习以及Java实现

Posted 2020-10-12

 

在计算机领域中，生产者-消费者问题（也叫bounded-buffer问题）是一类很经典的多进程同步问题。该问题描述了两类进程，即生产者进程和消费者进程，它们共享一个固定大小的缓冲区作为队列。生产者的任务是产生数据，并放到缓冲区中；同时消费者会消费数据（或者说将数据从缓冲区中移走）。每次生产或者消费数据的个数都是一个。在解决该问题时要确保当缓冲区满了之后，不会再向其中添加数据；当缓冲区为空的时候，消费者不能从中提取数据。

 

当缓冲区满了时，生产者要么进入休眠状态，要么丢弃产生的数据。等到下次消费者从缓冲区中移走一项数据之后，要通知生产者继续往缓冲区中添加数据。同样的，当消费者发现缓冲区为空时，消费者进入休眠状态。当生产者往缓冲区中写入数据之后，生产者要唤醒消费者来消费数据。该问题可以通过进程间通信来解决，典型的做法是使用信号量。如果该问题解决不当会导致死锁，即生产者和消费者都处于等待唤醒的状态。

 

这里参考了网上的写法，写了一个很简单的版本。有4个文件：

1. 容器，Container.java

2. 生产者，Producer.java

3. 消费者，Consumer.java

4. 测试文件，Test.java

 

Producer.java

package com.tuhooo.practice.pcmodel;

/**
 * 生产者
 */
public class Producer implements Runnable {

    private Container<Integer> container;
    private Object producerMonitor;
    private Object consumerMonitor;

    Producer(Container<Integer> container, Object producerMonitor, Object consumerMonitor) {
        this.container = container;
        this.producerMonitor = producerMonitor;/*生产者的锁*/
        this.consumerMonitor = consumerMonitor;/*消费者的锁*/
    }

    public void run() {
        while (true) {
            produce();
        }
    }

    private void produce() {

        if (container.isFull()) {
            synchronized (producerMonitor) {
                try {
                    if (container.isFull()) {
                        producerMonitor.wait();
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        } else {
            synchronized (producerMonitor) {
                if (!container.isFull()) {
                    container.add(0);
                    try {
                        Thread.currentThread().sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

                    System.out.println("生产者【" + Thread.currentThread().getId() + "】生产了一个, 一共" + container.getSize() + "个");
                }
            }
        }

        if (!container.isEmpty()) {
            synchronized (consumerMonitor) {
                if (!container.isEmpty()) {
                    consumerMonitor.notify();
                }
            }
        }
    }
}

Consumer.java

package com.tuhooo.practice.pcmodel;

/**
 * 消费者
 */
public class Consumer implements Runnable {

    private Container<Integer> container;
    private Object producerMonitor;
    private Object consumerMonitor;

    Consumer(Container<Integer> container, Object producerMonitor, Object consumerMonitor) {
        this.container = container;
        this.producerMonitor = producerMonitor;/*生产者的锁*/
        this.consumerMonitor = consumerMonitor;/*消费者的锁*/
    }
    
    public void run() {
        while (true) {
            consume();
        }
    }

    /*消费方法怎么写*/
    private void consume() {

        if (container.isEmpty()) {
            //消费者挂起
            synchronized (consumerMonitor) {
                try {
                    if (container.isEmpty()) {
                        consumerMonitor.wait();
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        } else {
            synchronized (consumerMonitor) {
                if (!container.isEmpty()) {
                    container.get();
                    try {
                        Thread.currentThread().sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

                    System.out.println("消费者【" + Thread.currentThread().getId() + "】消费了一个, 还有" + container.getSize() + "个");
                }
            }
        }

        if (!container.isFull()) {
            synchronized (producerMonitor) {
                if (!container.isFull()) {
                    producerMonitor.notify();
                }
            }
        }
    }
}

Container.java

package com.tuhooo.practice.pcmodel;

import java.util.LinkedList;
import java.util.List;

/**
 * 容器
 */
public class Container<T> {

    private int capacity;
    private List<T> list;

    public Container(int capacity) {
        this.capacity = capacity;
        list = new LinkedList<T>();
    }

    /*往容器中添加*/
    public synchronized boolean add(T product) {
        if(list.size() < capacity) {
            list.add(product);
            return true;
        }
        return false;
    }

    /*从容器中取*/
    public synchronized T get() {
        if(list.size() > 0) {
            list.remove(0);
        }
        return null;
    }

    /*判断是否是满的*/
    public boolean isFull() {
        return list.size() >= capacity;
    }

    public boolean isEmpty() {
        return list.size() == 0;
    }

    public synchronized int getSize() {
        return list.size();
    }

    public int getCapacity() {
        return this.capacity;
    }
}

Container有点坑的地方就是不知道什么时候应该加synchronized

 

Test.java

package com.tuhooo.practice.pcmodel;

public class Test {

    public static void main(String[] args) {

        Container<Integer> container = new Container<Integer>(10);
        Object producerMonitor = new Object();
        Object consumerMonitor = new Object();
        
        Thread p1 = new Thread(new Producer(container, producerMonitor, consumerMonitor));
        Thread p2 = new Thread(new Producer(container, producerMonitor, consumerMonitor));
        Thread c1 = new Thread(new Consumer(container, producerMonitor, consumerMonitor));
        Thread c2 = new Thread(new Consumer(container, producerMonitor, consumerMonitor));

        p1.start();
        p2.start();
        c1.start();
        c2.start();
    }
}

 

目前先就这么写，暂时运行起来没啥问题，如果有问题恳请各位看官指出。

 

有待改进地方：

1. Container有没有更好的实现方法

2. Container可以有多个，这样的话可以随机找到一个可用的Container分配给Producer或者Consumer用

3. 感觉代码中有好多的synchronized，感觉性能不是很好