自定义Lock和Condition实现生产消费模式

小灰灰 2023-07-25 11:13 7阅读 0赞

package com.example.test;
    
    import java.util.Queue;
    import java.util.Random;
    import java.util.concurrent.ConcurrentLinkedQueue;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.locks.AbstractQueuedSynchronizer;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    
    public class TestAqs implements Lock {
    
        public class Aqs extends  AbstractQueuedSynchronizer{
            //加锁
            @Override
            protected boolean tryAcquire(int arg) {
               if(compareAndSetState(0, arg)){
                   setExclusiveOwnerThread(Thread.currentThread());
                   return true;
               }
               return  false;
            }
            //释放锁
            @Override
            protected boolean tryRelease(int arg) {
                //判断当前state是否为0，为0不持有当前对象锁则抛出异常
                if(getState() == 0){
                    throw new IllegalMonitorStateException();
                }
                setState(0);
                setExclusiveOwnerThread(null);
                return true;
            }
    
            //判断当前锁是否被持有
            @Override
            protected boolean isHeldExclusively() {
                return  this.getState() == 1;
            }
    
            public Condition newCondition(){
                return new ConditionObject();
            }
        }
        //创建实例对象，修改state的值来实现加锁释放锁
        Aqs aqs = new Aqs();
        @Override
        public void lock() {
            aqs.acquire(1);
        }
    
        @Override
        public void lockInterruptibly() throws InterruptedException {
          aqs.acquireInterruptibly(1);
        }
    
        @Override
        public boolean tryLock() {
            return aqs.tryAcquire(1);
        }
    
        @Override
        public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
            return aqs.tryAcquireNanos(1, unit.toNanos(time));
        }
    
        @Override
        public void unlock() {
            aqs.release(0);
        }
    
        @Override
        public Condition newCondition() {
            return aqs.newCondition();
        }
    
        public static final TestAqs testAqs = new TestAqs();
        public static final Condition empty = testAqs.newCondition();
        public static final Condition full = testAqs.newCondition();
        public static final Queue queue = new ConcurrentLinkedQueue();
        public static final  int max = 10;
        public static final Random random = new Random();
        public static void main(String[] args){
            //生产线程
            Thread producer = new Thread(() ->{
                    while(true){
                        //获取独占锁
                        testAqs.lock();
                        try {
                            while(queue.size() == max){
                                System.out.println("队列饱和");
                                full.await();
                            }
                            add(queue);
                            empty.signalAll();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        } finally {
                            testAqs.unlock();
                        }
                    }
            });
    
            //消费线程
            Thread consumer = new Thread(() -> {
                    while (true) {
                        testAqs.lock();
                        System.out.println("消费者消费");
                        try {
                            while(queue.size() == 0){
                                empty.await();
                            }
                            reduce(queue);
                            full.signalAll();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        } finally {
                            testAqs.unlock();
                        }
                    }
            });
    
            producer.start();
            consumer.start();
    
        }
    
        public static void add( Queue queue){
                int value = random.nextInt(100);
                queue.add(value);
                System.out.println("队列添加元素：" + value);
        }
    
    
        public static void reduce(Queue queue){
                Object value = queue.poll();
                System.out.println("队列取出元素" + value);
        }
    }