6.30 生产者消费者

1. 使用BlockingQueue

public class ProducerConsumer
{

    public static void main(String[] args) throws InterruptedException
    {
        SynchronousQueue<Integer> queue = new SynchronousQueue<>();
        Producer producer = new Producer(queue);
        Consumer consumer = new Consumer(queue);

        Thread thread1 = new Thread(producer);
        Thread thread2 = new Thread(consumer);

        thread1.start();
        thread2.start();
        thread1.join();
        thread2.join();
    }
}


class Producer implements Runnable
{

    private SynchronousQueue<Integer> queue;

    public Producer(SynchronousQueue<Integer> queue)
    {
        this.queue = queue;
    }

    @Override
    public void run()
    {
        for (int i = 0; i < 10000; i++)
        {
            try
            {
                TimeUnit.SECONDS.sleep(1);
                System.out.println("生产者：" + i);
                queue.put(i);
            }
            catch (InterruptedException e)
            {
                e.printStackTrace();
            }

        }
    }
}


class Consumer implements Runnable
{
    private SynchronousQueue<Integer> queue;

    public Consumer(SynchronousQueue<Integer> queue)
    {
        this.queue = queue;
    }
    @Override
    public void run()
    {
        while (true)
        {
            try
            {
                Integer val = queue.take();
                System.out.println("消费者：" + val);

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

2. 使用wait notify

public class ProducerConsumer2
{

    static class Producer implements Runnable
    {
        private final List<Integer> queue;
        private final int fullSize;
        private final Object lock;

        public Producer(List<Integer> queue, int fullSize, Object lock)
        {
            this.queue = queue;
            this.fullSize = fullSize;
            this.lock = lock;
        }

        @Override
        public void run()
        {

            for (int i = 0; i < 10000000; i++)
            {
                synchronized (lock)
                {
                    while (queue.size() == fullSize)
                    {
                        try
                        {
                            lock.wait();
                        }
                        catch (InterruptedException e)
                        {
                            e.printStackTrace();
                        }
                    }
                    System.out.println(Thread.currentThread().getName() + " put val: " + i);
                    queue.add(i);
                    lock.notifyAll();
                }

            }
        }
    }

    static class Consumer implements Runnable
    {
        private final List<Integer> queue;
        private final int fullSize;
        private final Object lock;

        public Consumer(List<Integer> queue, int fullSize, Object lock)
        {
            this.queue = queue;
            this.fullSize = fullSize;
            this.lock = lock;
        }

        @Override
        public void run()
        {

            while (true)
            {
                synchronized (lock)
                {
                    while (queue.isEmpty())
                    {
                        try
                        {
                            lock.wait();
                        }
                        catch (InterruptedException e)
                        {
                            e.printStackTrace();
                        }
                    }

                    Integer val = queue.remove(0);
                    System.out.println(Thread.currentThread().getName() + " get val: " + val);

                    lock.notifyAll();
                }
            }
        }
    }


    public static void main(String[] args) throws InterruptedException
    {
        final List<Integer> queue = new ArrayList<>();
        final int fullSize = 5;
        final Object lock = ProducerConsumer2.class;
        Producer producer = new Producer(queue, fullSize, lock);
        Consumer consumer = new Consumer(queue, fullSize, lock);

        Thread thread1 = new Thread(producer, "producer");
        Thread thread2 = new Thread(consumer, "consumer");


        thread1.start();
        thread2.start();
        thread1.join();
        thread2.join();

    }
}

3. 使用Lock Condition

相比于上面的wait notify这里用了两个condition，这样子唤醒的时候就不会把生产者和消费者一起唤醒，只唤醒某一个即可（即生产者唤醒消费者，消费者唤醒生产者）

public class ConditionTest
{
    private Lock lock;//一个锁说明读写互斥
    private int capacity;
    private List<Object> items;
    private Condition notFull;//用来唤醒写线程
    private Condition notEmpty;//用来唤醒读线程

    public ConditionTest(int capacity)
    {
        this.capacity = capacity;
        this.items = new ArrayList<>();
        this.lock = new ReentrantLock();
        this.notFull = lock.newCondition();
        this.notEmpty = lock.newCondition();
    }

    public void add(Object data) throws InterruptedException
    {
        try
        {
            lock.lock();
            //新增的时候如果已经满了，那么等待 非满信号 唤醒
            while (this.items.size() == capacity)
            {
                this.notFull.await();
            }

            //增加了一个元素，那么 唤醒非空
            this.items.add(data);
            this.notEmpty.signalAll();

        }
        finally
        {
            lock.unlock();
        }
    }

    public Object remove() throws InterruptedException
    {
        try
        {
            lock.lock();

            //删除的时候已经空了，那么等待 非空信号 唤醒
            while (this.items.size() == 0)
            {
                this.notEmpty.await();
            }

            //删除了一个元素，那么 唤醒非满
            Object data = this.items.remove(0);
            this.notFull.signalAll();
            return data;
        }
        finally
        {
            lock.unlock();
        }
    }

    public static void main(String[] args)
    {
        ConditionTest conditionTest = new ConditionTest(5);
        new Thread(() -> {
            for (int i = 0; i < 1000; i++)
            {
                try
                {
                    conditionTest.add(i);
                    System.out.println(String.format("生产者放入%d", i));
                }
                catch (InterruptedException e)
                {
                    e.printStackTrace();
                }
            }
        }).start();

        new Thread(() -> {
            try
            {
                while (true)
                {
                    Object data = conditionTest.remove();
                    System.out.println(String.format("消费者消费%d", data));
                }
            }
            catch (InterruptedException e)
            {
                e.printStackTrace();
            }
        }).start();



    }
}