怎么实现实现一个生产者消费者？

### 1.使用Object.wait()和Object.notify()实现 使用queue作为一个队列，存放数据，并且使用Synchronized同步锁，每次只能同时存在一个线程来生产或者消费数据， 生成线程发现队列容量>10,生产者线程就进入waiting状态，一旦成功往队列添加数据，那么就唤醒所有线程（主要是生产者线程起来消费）。 消费线程消费时，发现队列容量==0，也会主动进入waiting状态。 伪代码如下： ```java LinkedList<Integer> queue = new LinkedList<>(); void produce(Integer value) { synchronized(queue) {//加锁控制，保证同一时间点，只能有一个线程生成或者消费 while(queue.size()>10) { queue.waiting(); } queue.add(value); //唤醒消费者线程 queue.notifyAll(); } } Integer consumer() { synchronized(queue) {//加锁控制，保证同一时间点，只能有一个线程生成或者消费 while(queue.size()==0) { queue.waiting(); } Integer value = queue.poll(); //唤醒生产者线程 queue.notifyAll(); return value; } } ``` 完整代码如下： ```java public static void main(String[] args) { Queue<Integer> queue = new LinkedList<>(); final Customer customer = new Customer(queue); final Producer producer = new Producer(queue); ExecutorService pool = Executors.newCachedThreadPool(); for (int i = 0; i < 1000; i++) { pool.execute(new Runnable() { @Override public void run() { try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } Integer a = customer.removeObject(); System.out.println("消费了数据 "+a); } }); pool.execute(new Runnable() { @Override public void run() { try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } Random random = new Random(); Integer a = random.nextInt(1000); System.out.println("生成了数据 "+a); producer.addObject(a); } }); } } private static class Customer { Queue<Integer> queue; Customer(Queue<Integer> queue) { this.queue = queue; } public Integer removeObject() { synchronized (queue) { try { while (queue.size()==0) { System.out.println("队列中没有元素了，进行等待"); queue.wait(); } } catch (InterruptedException e) { e.printStackTrace(); } Integer number = queue.poll(); System.out.println("唤醒所有生产线程，当前queue大小是" + queue.size()); queue.notifyAll(); return number; } } } private static class Producer { Queue<Integer> queue; Producer(Queue<Integer> queue) { this.queue = queue; } public void addObject(Integer number) { synchronized (queue) { try { while (queue.size()>10) { queue.wait(); } } catch (InterruptedException e) { e.printStackTrace(); } queue.add(number); queue.notifyAll(); System.out.println("唤醒所有消费线程，当前queue大小是"+queue.size()); } } } ``` ### 2.使用Lock和Condition来实现 调用Object.wait()方法可以让线程进入等待状态，被添加到Object的monitor监视器的等待队列中，Object.notifyAll()可以唤醒monitor监视器等待队列中的所有线程。 而调用lock的newCondition()方法,可以返回一个ConditionObject实例对象，每个ConditionObject包含一个链表，存储等待队列。可以认为一个ReentrantLock有一个同步队列（存放没有获得锁的线程），和多个等待队列（存放调用await()方法的线程）。使用Condition.singal()和Condition.singalAll()可以更加精准的唤醒线程，也就是唤醒的都是这个Condition对应的等待队列里面的线程，而Object.notify()和Object.notifyAll()只能唤醒等待队列中的所有的线程。 ```java ReentrantLock lock = new ReentrantLock(); Condition customerQueue = lock.newCondition(); ``` ReentrantLock的Condition相关的实现  ```java abstract static class Sync extends AbstractQueuedSynchronizer { final ConditionObject newCondition() { return new ConditionObject(); } } //AQS内部类 ConditionObject public class ConditionObject implements Condition, java.io.Serializable { private static final long serialVersionUID = 1173984872572414699L; //链表头结点 private transient Node firstWaiter; //链表尾结点 private transient Node lastWaiter; //真正的创建Condition对象 public ConditionObject() { } } ``` 消费者-生产者实现 ```java public static void main(String[] args) { ReentrantLock lock = new ReentrantLock(); Condition customerQueue = lock.newCondition(); Condition producerQueue = lock.newCondition(); Queue<Integer> queue = new LinkedList<>(); final Customer customer = new Customer(lock,customerQueue, producerQueue,queue); final Producer producer = new Producer(lock,customerQueue, producerQueue,queue); ExecutorService pool = Executors.newCachedThreadPool(); for (int i = 0; i < 1000; i++) { pool.execute(new Runnable() { @Override public void run() { try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } Integer a = customer.take(); // System.out.println("消费了数据 "+a); } }); pool.execute(new Runnable() { @Override public void run() { try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } Random random = new Random(); Integer a = random.nextInt(1000); // System.out.println("生成了数据 "+a); producer.add(a); } }); } } private static class Customer { private ReentrantLock lock; private Condition customer; private Condition producer; private Queue<Integer> queue; Customer(ReentrantLock lock, Condition customer, Condition producer,Queue<Integer> queue) { this.lock = lock; this.customer = customer; this.producer = producer; this.queue = queue; } public Integer take() { lock.lock(); Integer element = null; try { while (queue.size() == 0) { customer.await(); } element = queue.poll(); System.out.println("消费者线程取出来元素"+element); producer.signalAll(); } catch (InterruptedException e) { e.printStackTrace(); } finally { lock.unlock(); } return element; } } private static class Producer { private ReentrantLock lock; private Condition customer; private Condition producer; private Queue<Integer> queue; Producer(ReentrantLock lock, Condition customer, Condition producer,Queue<Integer> queue) { this.lock = lock; this.customer = customer; this.producer = producer; this.queue = queue; } public void add( Integer element) { lock.lock(); try { while (queue.size() > 10) { producer.await(); } queue.add(element); System.out.println("生成和线程添加元素"+element); customer.signalAll(); } catch (InterruptedException e) { e.printStackTrace(); } finally { lock.unlock(); } } } ``` ### 3.使用BlockingQueue实现 利用阻塞队列BlockingQueue的特征进行生产和消费的同步(其实阻塞队列内部也是基于Lock,condition实现的 ) ```java public class BlockQueueRepository<T> extends AbstractRepository<T> implements Repository<T> { public BlockQueueRepository(int cap) { //cap代表队列的最大容量 products = new LinkedBlockingQueue<>(cap); } @Override public void put(T t) { if (isFull()) { log.info("repository is full, waiting for consume....."); } try { //如果队列长度已满，那么会阻塞等待 ((BlockingQueue) products).put(t); } catch (InterruptedException e) { e.printStackTrace(); } } @Override public T take() { T product = null; if (isEmpty()) { log.info("repository is empty, waiting for produce....."); } try { //如果队列元素为空，那么也会阻塞等待 product = (T) ((BlockingQueue) products).take(); } catch (InterruptedException e) { e.printStackTrace(); } return product; } } ```