Java多线程（6）：锁与AQS（下）

技术分享 3年前 (2022-10-31) 0 999+

您好，我是湘王，这是我的博客园，欢迎您来，欢迎您再来～

之前说过，AQS（抽象队列同步器）是Java锁机制的底层实现。既然它这么优秀，是骡子是马，就拉出来溜溜吧。

首先用重入锁来实现简单的累加，就像这样：

/**  * 用重入锁实现累加  *  * @author 湘王  */ public class MyLockTest {     private final Lock lock = new ReentrantLock();     private int value;     public int getNext() {         lock.lock();         try {             value++;         } finally {             lock.unlock();         }         return value;     }     public static void main(String[] args) {         MyLockTest myLock = new MyLockTest();         for (int i = 0; i < 5; i++) {             new Thread(new Runnable() {                 @Override                 public void run() {                     for (int i = 0; i < 5; i++) {                         System.out.println(myLock.getNext());                     }                 }             }).start();         }     } }

运行结果显示数据有重复：

这么简单的计算都能出现重复，这肯定是无法接受的。

再用独占锁来试试看：

/**  * 利用AQS实现自定义独占锁  *  * @author 湘王  */ public class MyExclusiveLock implements Lock {     @Override     public void lock() {      }      @Override     public void lockInterruptibly() throws InterruptedException {      }      @Override     public boolean tryLock() {         return false;     }      @Override     public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {         return false;     }      @Override     public void unlock() {      }      @Override     public Condition newCondition() {         return null;     } }

可以看到，实现lock接口，就需要实现若干自定义的接口。然后以内部类继承AQS的方式，实现排他锁，昨天也说过，AQS中tryAcquire()和tryRelease()是一一对应的，也就是也管获取，一个管释放，所以代码是：

/**  * 内部类继承AQS的方式，实现排他锁  */ private static class SyncHelper extends AbstractQueuedSynchronizer {     private static final long serialVersionUID = -7666580981453962426L;      /**      * 第一个线程进来，拿到锁就返回true；后面的线程进来，拿不到锁就返回false      */     @Override     protected boolean tryAcquire(int arg) {         // 获取资源状态         int state = getState();         if (0 == state) {// 如果没有线程拿到资源的锁             if (compareAndSetState(0, arg)) {                 // 保存当前持有同步锁的线程                 setExclusiveOwnerThread(Thread.currentThread());                 return true;             }         } else if (Thread.currentThread() == getExclusiveOwnerThread()) {             // 如果当前线程再次进来，state + 1，可重入             // 如果这里没有这个判断，那么程序会卡死             setState(state + arg);             return true;         }         return false;     }      /**      * 锁的获取和释放需要一一对应      */     @Override     protected boolean tryRelease(int arg) {         // 获取资源状态         int state = getState();         // 返回最后一个通过setExclusiveOwnerThread()方法设置过的线程，或者null         if (Thread.currentThread() != getExclusiveOwnerThread()) {             throw new RuntimeException();         }         setState(state - arg);         if (0 == state) {             setExclusiveOwnerThread(null);             return true;         }         return false;     }      protected Condition newCondition() {         return new ConditionObject();     } }

然后再用AQS实现lock接口的方法：

/**  * 利用AQS实现自定义独占锁  *  * @author 湘王  */ public class MyExclusiveLock implements Lock {     private final SyncHelper synchepler = new SyncHelper();      @Override     public void lock() {         synchepler.acquire(1);     }      @Override     public void lockInterruptibly() throws InterruptedException {         synchepler.acquireInterruptibly(1);     }      @Override     public boolean tryLock() {         return synchepler.tryAcquire(1);     }      @Override     public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {         return synchepler.tryAcquireNanos(1, unit.toNanos(time));     }      @Override     public void unlock() {         synchepler.release(1);     }      @Override     public Condition newCondition() {         return synchepler.newCondition();     }      /**      * 内部类继承AQS的方式，实现排他锁      */     private static class SyncHelper extends AbstractQueuedSynchronizer {         private static final long serialVersionUID = -7666580981453962426L;              /**          * 第一个线程进来，拿到锁就返回true；后面的线程进来，拿不到锁就返回false          */         @Override         protected boolean tryAcquire(int arg) {             // 获取资源状态             int state = getState();             if (0 == state) {// 如果没有线程拿到资源的锁                 if (compareAndSetState(0, arg)) {                     // 保存当前持有同步锁的线程                     setExclusiveOwnerThread(Thread.currentThread());                     return true;                 }             } else if (Thread.currentThread() == getExclusiveOwnerThread()) {                 // 如果当前线程再次进来，state + 1，可重入                 // 如果这里没有这个判断，那么程序会卡死                 setState(state + arg);                 return true;             }             return false;         }              /**          * 锁的获取和释放需要一一对应          */         @Override         protected boolean tryRelease(int arg) {             // 获取资源状态             int state = getState();             // 返回最后一个通过setExclusiveOwnerThread()方法设置过的线程，或者null             if (Thread.currentThread() != getExclusiveOwnerThread()) {                 throw new RuntimeException();             }             setState(state - arg);             if (0 == state) {                 setExclusiveOwnerThread(null);                 return true;             }             return false;         }              protected Condition newCondition() {             return new ConditionObject();         }     } }

然后再运行测试：

/**  * 实现Lock接口方法并运行排他锁测试  *  * @author 湘王  */ public class MyExclusiveLockTester {     // 用自定义AQS独占锁实现     private Lock lock = new MyExclusiveLock();     private int value;      public int accmulator() {         lock.lock();         try {             ++value;         } finally {             lock.unlock();         }          return value;     }      public static void main(String[] args) throws InterruptedException {         MyExclusiveLockTester test = new MyExclusiveLockTester();         for (int i = 0; i < 5; i++) {             new Thread(new Runnable() {                 @Override                 public void run() {                     for (int i = 0; i < 5; i++) {                         System.out.println(test.accmulator());                     }                 }             }).start();         }     } }

可以看到，结果无论怎么样都不会再重复了。

这个只是简单的累加，接下来用AQS来实现一个实际的生活场景。比如周末带女票或男票去步行街吃饭，这时候人特别多，需要摇号，而且一次只能进去三张号（不按人头算，按叫到的号来算），该怎么实现呢？

可以顺着这个思路：摇号机虽有很多号，但它本质上是个共享资源，很多人可以共享，但是每次共享的数量有限。这其实就是个可以指定数量的共享锁而已。

既然有了思路，那接下来就好办了。

/**  * 利用AQS实现自定义共享锁  *  * @author 湘王  */ public class MyShareLock implements Lock {     @Override     public void lock() {     }      @Override     public void lockInterruptibly() throws InterruptedException {     }      @Override     public boolean tryLock() {         return false;     }      @Override     public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {         return false;     }      @Override     public void unlock() {     }      @Override     public Condition newCondition() {         return null;     } }

还是一样实现Lock接口，但这次是用AQS实现共享锁。

/**  * 内部类继承AQS实现共享锁  *  */ private static class SyncHelper extends AbstractQueuedSynchronizer {     private static final long serialVersionUID = -7357716912664213942L;      /**      * count表示允许几个线程能同时获得锁      */     public SyncHelper(int count) {         if (count <= 0) {             throw new IllegalArgumentException("锁资源数量必须大于0");         }         // 设置资源总数         setState(count);     }      /**      * 一次允许多少个线程进来，允许数量的线程都能拿到锁，其他的线程进入队列      */     @Override     protected int tryAcquireShared(int acquires) {         // 自旋         for (;;) {             int state = getState();             int remain = state - acquires;             // 判断剩余锁资源是否已小于0或者CAS执行是否成功             if (remain < 0 || compareAndSetState(state, remain)) {                 return remain;             }         }     }      /**      * 锁资源的获取和释放要一一对应      */     @Override     protected boolean tryReleaseShared(int releases) {         // 自旋         for (;;) {             // 获取当前state             int current = getState();             // 释放状态state增加releases             int next = current + releases;             if (next < current) {// 溢出                 throw new Error("Maximum permit count exceeded");             }             // 通过CAS更新state的值             // 这里不能用setState()             if (compareAndSetState(current, next)) {                 return true;             }         }     }      protected Condition newCondition() {         return new ConditionObject();     } }

然后再来改造之前实现的接口：

/**  * 利用AQS实现自定义共享锁  *  * @author 湘王  */ public class MyShareLock implements Lock {     public static int count;     private final SyncHelper synchepler = new SyncHelper(count);      @Override     public void lock() {         synchepler.acquireShared(1);     }      @Override     public void lockInterruptibly() throws InterruptedException {         synchepler.acquireSharedInterruptibly(1);     }      @Override     public boolean tryLock() {         return synchepler.tryAcquireShared(1) > 0;     }      @Override     public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {         return synchepler.tryAcquireSharedNanos(1, unit.toNanos(time));     }      @Override     public void unlock() {         synchepler.releaseShared(1);     }      @Override     public Condition newCondition() {         return synchepler.newCondition();     }      /**      * 内部类继承AQS实现共享锁      *      */     private static class SyncHelper extends AbstractQueuedSynchronizer {         private static final long serialVersionUID = -7357716912664213942L;          /**          * count表示允许几个线程能同时获得锁          */         public SyncHelper(int count) {             if (count <= 0) {                 throw new IllegalArgumentException("锁资源数量必须大于0");             }             // 设置资源总数             setState(count);         }          /**          * 一次允许多少个线程进来，允许数量的线程都能拿到锁，其他的线程进入队列          */         @Override         protected int tryAcquireShared(int acquires) {             // 自旋             for (;;) {                 int state = getState();                 int remain = state - acquires;                 // 判断剩余锁资源是否已小于0或者CAS执行是否成功                 if (remain < 0 || compareAndSetState(state, remain)) {                     return remain;                 }             }         }          /**          * 锁资源的获取和释放要一一对应          */         @Override         protected boolean tryReleaseShared(int releases) {             // 自旋             for (;;) {                 // 获取当前state                 int current = getState();                 // 释放状态state增加releases                 int next = current + releases;                 if (next < current) {// 溢出                     throw new Error("Maximum permit count exceeded");                 }                 // 通过CAS更新state的值                 // 这里不能用setState()                 if (compareAndSetState(current, next)) {                     return true;                 }             }         }          protected Condition newCondition() {             return new ConditionObject();         }     } }

接下来就该测试咱们需要的效果是否能实现了：

public class MyShareLockTester {     public static void main(String[] args) throws InterruptedException {         // 用自定义AQS共享锁实现         // 一次允许发放三把锁         MyShareLock.count = 3;         final Lock lock = new MyShareLock();          // 模拟20个客户端访问         for (int i = 0; i < 20; i++) {             new Thread(new Runnable() {                 @Override                 public void run() {                     try {                         lock.lock();                         System.out.println("持有 " + Thread.currentThread().getName() + " 的客人可以进餐厅就餐");                         // 每两次叫号之间间隔一段时间，模拟真实场景                         Thread.sleep(3000);                     } catch (InterruptedException e) {                         e.printStackTrace();                     } finally {                         // 使用完成释放锁                         lock.unlock();                     }                 }             }).start();         }     } }