AQS的应用：基于AQS实现自定义同步器

骑猪看日落 2022-12-30 08:23 99阅读 0赞

> 微信公众号：Zhongger  
> 我是Zhongger，一个在互联网行业摸鱼写代码的打工人！  
> 关注我，了解更多你不知道的【Java后端】打工技巧、职场经验、生活感悟等

### 三、基于AQS实现自定义同步器 ###

之前学习了这么多关于AQS的原理性的知识，这一期，我们来基于AQS实现一个不可重入的独占锁， 自定义AQS需要重写一系列函数，还需要定义原子变量state的含义。这里我们定义， **state=0 表示目前锁没有被线程持有** ，**state=1 表示锁己经被某一个线程持有**。 由于是不可重入锁，所以不需要记录持有锁的线程获取锁的次数。另外，还要自定义锁的支持条件变量。

#### 1、代码实现 ####

如下代码是基于AQS实现不可重入的独占锁。

package MyNonReentrantLock;
    
    import java.io.Serializable;
    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 NonReentrantLock implements Lock, Serializable { 
        //静态内部类，用于辅助
        private static class Sync extends AbstractQueuedSynchronizer{ 
            @Override
            protected boolean tryAcquire(int arg) { 
               assert arg == 1;//如果state为0，则尝试获取锁
                if (compareAndSetState(0,1)){ 
                    setExclusiveOwnerThread(Thread.currentThread());
                    return true;
                }
                return false;
            }
    
            @Override
            protected boolean tryRelease(int arg) { 
                assert arg == 1;//如果state为0，则尝试获取锁
                if (getState()==0){ 
                    throw new IllegalMonitorStateException();
                }
                setExclusiveOwnerThread(null);
                setState(0);
                return true;
            }
    
            @Override
            protected boolean isHeldExclusively() { 
                // 是否锁已经被持有
                return getState()==1;
            }
    
            //提供条件变量接口
            public Condition newCondition(){ 
                return new ConditionObject();
            }
        }
    
        Sync sync = new Sync();
    
        @Override
        public void lock() { 
            sync.acquire(1);
        }
    
        @Override
        public void lockInterruptibly() throws InterruptedException { 
            sync.acquireInterruptibly(1);
        }
    
        @Override
        public boolean tryLock() { 
            return sync.tryAcquire(1);
        }
    
        @Override
        public boolean tryLock(long time, TimeUnit unit) throws InterruptedException { 
            return sync.tryAcquireNanos(1, unit.toNanos(time));
        }
    
        @Override
        public void unlock() { 
            sync.release(1);
        }
    
        @Override
        public Condition newCondition() { 
            return sync.newCondition();
        }
    }

在如上代码中， NonReentrantLock 定义了一个内部类 Sync 用来实现具体的锁的操作， Sync 继承了AQS 。由于实现的是独占模式的锁，所以Sync重写了`tryAcquire/tryRelease/isHeldExclusively`个方法。另 外， Sync 提供 `newCondition` 这个方法用来支持条件变量。

#### 2、使用自定义锁实现生产一消费模型 ####

代码如下：

package MyNonReentrantLock;
    
    import java.util.Queue;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.locks.Condition;
    
    public class 自定义生产消费模型 { 
        static NonReentrantLock lock = new NonReentrantLock();
        static Condition notFull = lock.newCondition();
        static Condition notEmpty = lock.newCondition();
        static Queue<String> queue = new LinkedBlockingQueue<>();
        static int queueSize = 10;
    
        public static void main(String[] args) { 
            Thread producer = new Thread(() -> { 
                lock.lock();
                try { 
                    //如果队列满了，则等待
                    while (queue.size() == queueSize) { 
                        notEmpty.await();
                    }
                    //添加队列元素
                    queue.add("element ");
                    //唤醒消费线程
                    notFull.signalAll();
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }finally { 
                    //释放锁
                    lock.unlock();
                }
    
            });
    
            Thread consumer = new Thread(() -> { 
                lock.lock();
    
                try { 
                    //队列为空，则等待
                    while (queue.size()==0){ 
                        notFull.await();
                    }
                    //消费元素
                    queue.poll();
                    //唤醒生产线程
                    notEmpty.signalAll();
    
                }catch (InterruptedException e){ 
                    e.printStackTrace();
                } finally { 
                    lock.unlock();
                }
            });
            producer.start();
            consumer.start();
        }
    }

如上代码首先创建了NonReentrantLock的一个对象lock ，然后调用 lock.newCondition创建了两个条件变量，用来进行生产者和消费者线程之间的同步。

在 main 函数里面，首先创建了 producer 生产线程，在线程内部首先调用 lock.lock()获取独占锁，然后判断当前队列是否己经满了 ，如果满了则调用 notEmpty.await（）阻塞挂起当前线程。需要注意的是，这里使用 while 而不是 if 是为了**避免虚假唤醒** 。如果队列不满则直接向队列里面添加元素，然后调用 notFull.signalAll唤醒所有因为消费元素而被阻塞的消费线程，最后释放获取的锁。

然后在 main 函数里面创建了 consumer 生产线程，在线程内部首先调用 lock.lock()获取独占锁，然后判断当前队列里面是不是有元素，如果队列为空则调用 notFull.await（）阻塞挂起当前线程。需要注意的是，这里使用 while 而不 if 是为了**避免虚假唤醒**。如果队列不为空则直接从队列里面获取并移除元素，然后唤醒因为队列满而被阻塞的生产线程，最后释放获取的锁。

到目前为止，AQS相关的知识就告一段落，后续的多线程学习中我们还是会继续看到AQS的影子！我是Zhongger，一个在互联网行业摸鱼写代码的打工人，你们的【关注】和【在看】与支持是我创作的最大动力，我们下期见~