Java并发编程之生产者消费者模式

àì夳堔傛蜴生んèń 2022-05-25 10:49 221阅读 0赞

有关Java多线程，前面我记录自己对于线程池创建的两种方式，以及线程池如何创建线程（这里自己在加深一下印象：当执行execute方法时候，会将任务放入addWorker方法中，addWorker将任务放入到Worker构造器中，该类中runWorker方法有一个getTask方法，这个方法中有一段代码：boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;  Runnable r = timed ?    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :  workQueue.take();会从队列中取出任务执行，否则take()核心线程进行阻塞，保证核心线程不被销毁）

[Java并发编程之线程池的使用（一）点击打开链接][Java]

[Java并发编程之线程池的使用（二）点击打开链接][Java 1]

[Java并发编程之线程池创建线程任务的过程  点击打开链接][Java 2]

在多线程并发任务中，很多并发问题是由于对于JVM中主内存（主要包括方法区和堆，有关JVM知识可以自行查阅）中共享变量的不安全操作。所以我觉得就扯出来锁，目前我所知道的就是乐观锁/悲观锁,公平锁/非公平锁，互斥锁/读写锁（ReentrantLock/ReadWriteLock）以及自旋锁，分段锁（concurrentHashMap）。这些锁在不同场景的应用能够解决不同的并发问题。我又想扯到AQS中，但是我理解还很肤浅，不在叙述，开始实现生产者消费者模式：

我这里用两种方法去实现：

一种是利用BlockingQueue,一种是利用Condition（多线程间协调通信工具类）的await、signalAll

package com.elastic.search.ProCus;
    
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    public class ProCus {
        public static void main(String[] args) {
            //new ProCus().BloQueueTest(); // BlockingQueue
            new ProCus().ConditionTest(); // await signalAll
        }
    
        public void BloQueueTest() {
            Queue queue = new Queue();
            new Thread(new Producer(queue), "pro1").start();
    
            new Thread(new Customer(queue), "c1").start();
            new Thread(new Customer(queue), "c2").start();
            new Thread(new Customer(queue), "c3").start();
        }
    
        public void ConditionTest() {
            Lock lock = new ReentrantLock();
            Condition producer = lock.newCondition();
            Condition customer = lock.newCondition();
            Cache cache = new Cache(lock, producer, customer);
            new Thread(new Produce2(cache), "p1").start();
    
            new Thread(new Customer2(cache), "c1").start();
            new Thread(new Customer2(cache), "c2").start();
        }
    }
    
    class Queue {
        private BlockingQueue blockingQueue = new LinkedBlockingQueue(10);
    
        public void Producer() {
            try {
                blockingQueue.put(1);
                System.out.println(Thread.currentThread().getName() + "生产，当前资源数：" + blockingQueue.size());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    
        public void customer() {
            try {
                System.out.println(Thread.currentThread().getName() + "进入到队列中准备消耗资源，此时资源池数量:" + blockingQueue.size());
                blockingQueue.take();
                System.out.println(Thread.currentThread().getName() + "消耗，当前资源数：" + blockingQueue.size());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    
    class Producer implements Runnable {
        private Queue queue;
    
        public Producer(Queue queue) {
            this.queue = queue;
        }
    
        @Override
        public void run() {
            for (; ; ) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                queue.Producer();
            }
        }
    }
    
    class Customer implements Runnable {
        private Queue queue;
    
        public Customer(Queue queue) {
            this.queue = queue;
        }
    
        @Override
        public void run() {
            for (; ; ) {
                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                queue.customer();
            }
        }
    }
    
    class Cache {
        int size = 10;
        int num = 0;
        private Lock lock;
        private Condition proCondition;
        private Condition cusCondition;
    
        Cache(Lock lock, Condition proCondition, Condition cusCondition) {
            this.lock = lock;
            this.cusCondition = cusCondition;
            this.proCondition = proCondition;
        }
    
        public void producer() {
            lock.lock();
            try {
                if (size > num) {
                    num++;
                    System.out.println(Thread.currentThread().getName() + "product,num:" + num);
                    cusCondition.signalAll();
                } else {
                    try {
                        System.out.println("size > num pro await");
                        proCondition.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            } finally {
                lock.unlock();
            }
        }
    
        public void customer() {
            lock.lock();
            try {
                if (num > 0) {
                    num--;
                    System.out.println(Thread.currentThread().getName() + "customer,num:" + num);
                    proCondition.signalAll();
                } else {
                    try {
                        System.out.println(" num  < 0 cus await");
                        cusCondition.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            } finally {
                lock.unlock();
            }
        }
    }
    
    class Produce2 implements Runnable {
        private Cache cache;
    
        Produce2(Cache cache) {
            this.cache = cache;
        }
    
        @Override
        public void run() {
            for (; ; ) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                cache.producer();
            }
        }
    }
    
    class Customer2 implements Runnable {
        private Cache cache;
    
        Customer2(Cache cache) {
            this.cache = cache;
        }
    
        @Override
        public void run() {
            for (; ; ) {
                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                cache.customer();
            }
        }
    }

[Java]: https://blog.csdn.net/qq442270636/article/details/80056801
[Java 1]: https://blog.csdn.net/qq442270636/article/details/80101671
[Java 2]: https://blog.csdn.net/qq442270636/article/details/80266524