Java 多线程 生产者&消费者 问题

Java 多线程 生产者&消费者 问题。

package cn.cnerp.demo1;

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class Demo {

    //

    long pspeed = 2000;
    long cspeed = 1000;

    BlockingQueue<String> linkedBlockingDeque = new LinkedBlockingDeque<>(10);

    Queue<String> queue = new LinkedList<>();
    Lock lock = new ReentrantLock();
    Condition producer = lock.newCondition();
    Condition consumer = lock.newCondition();


    public static void main(String[] args) {


        // BlockingQueue
        Demo demo = new Demo();
        ExecutorService executorService = Executors.newCachedThreadPool();
        executorService.submit(demo.new BlockingQueueProducer());
        executorService.submit(demo.new BlockingQueueConsumer());
        executorService.shutdown();

        // ReentrantLock
//        Demo demo = new Demo();
//        ExecutorService executorService = Executors.newCachedThreadPool();
//        executorService.submit(demo.new ReentrantLockProducer());
//        executorService.submit(demo.new ReentrantLockConsumer());
//        executorService.shutdown();

        // synchronized
//        Demo demo = new Demo();
//        ExecutorService executorService = Executors.newCachedThreadPool();
//        executorService.submit(demo.new SynchronizedProducer());
//        executorService.submit(demo.new SynchronizedConsumer());
//        executorService.shutdown();


    }


    class BlockingQueueProducer implements Runnable {

        @Override
        public void run() {

            while (true) {
                try {
                    linkedBlockingDeque.put("product:" + System.currentTimeMillis());
                    System.out.println("#Producer-" + Thread.currentThread().getName() + "# put one product , now size : " + linkedBlockingDeque.size());
                    Thread.sleep(pspeed);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }

        }

    }

    class BlockingQueueConsumer implements Runnable {
        @Override
        public void run() {

            while (true) {
                try {
                    linkedBlockingDeque.take();
                    System.out.println("#Consumer-" + Thread.currentThread().getName() + "# take one product , still left : " + linkedBlockingDeque.size());
                    Thread.sleep(cspeed);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }

        }
    }


    class ReentrantLockProducer implements Runnable {
        @Override
        public void run() {

            while (true) {

                try {

                    lock.lockInterruptibly();

                    while (queue.size() == 10) {
                        producer.await();
                    }

                    queue.add("product:" + System.currentTimeMillis());
                    consumer.signal();
                    System.out.println("#Producer-" + Thread.currentThread().getName() + "# put one product , now size : " + queue.size());
                    Thread.sleep(pspeed);

                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    lock.unlock();
                }

            }

        }
    }

    class ReentrantLockConsumer implements Runnable {
        @Override
        public void run() {

            while (true) {

                try {

                    lock.lockInterruptibly();

                    while (queue.size() == 0) {
                        consumer.await();
                    }

                    queue.poll();
                    producer.signal();
                    System.out.println("#Consumer-" + Thread.currentThread().getName() + "# take one product , still left : " + queue.size());
                    Thread.sleep(cspeed);

                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    lock.unlock();
                }

            }

        }
    }


    class SynchronizedProducer implements Runnable {
        @Override
        public void run() {

            while (true) {

                synchronized (queue) {

                    try {

                        while (queue.size() == 10) {
                            queue.wait();
                        }

                        queue.add("product:" + System.currentTimeMillis());
                        queue.notify();
                        System.out.println("#Producer-" + Thread.currentThread().getName() + "# put one product , now size : " + queue.size());
                        Thread.sleep(pspeed);

                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

                }

            }

        }
    }

    class SynchronizedConsumer implements Runnable {
        @Override
        public void run() {

            while (true) {

                synchronized (queue) {

                    try {

                        while (queue.size() == 0) {
                            queue.wait();
                        }

                        queue.poll();
                        queue.notify();
                        System.out.println("#Consumer-" + Thread.currentThread().getName() + "# take one product , still left : " + queue.size());
                        Thread.sleep(pspeed);

                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

                }

            }

        }
    }


}