Java中的Lock与AQS

梦里梦外; 2022-02-27 16:46 172阅读 0赞

锁是用来控制多个线程访问共享资源的方式。在Java SE 5之前，想要实现锁的功能只能使用synchronized，而在Java SE 5之后并发包中新增**了Lock接口以及相关实现类**用来实现锁的功能，它提供了与synchronized关键字类似的同步功能，只是在使用时需要显式地获取锁和释放锁。

### Lock接口 ###

**Lock的使用方式**

Lock lock = new ReentrantLock();
    lock.lock();
    try {
    } finally {
    	lock.unlock();
    }

在finally块中释放锁，目的是保证在获取到锁之后，最终能够被释放。

**Lock的API**

<table> 
 <thead> 
  <tr> 
   <th>方法名称</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>void lock()</td> 
   <td>获取锁，调用该方法当前线程将会获取锁</td> 
  </tr> 
  <tr> 
   <td>void lockInterruptibly() throws InterruptedException</td> 
   <td>如果当前线程未被中断且锁可用，则获取锁，并立即返回。与lock()方法的不同之处在于该方法会响应中断，即在锁的获取中可以中断当前线程。</td> 
  </tr> 
  <tr> 
   <td>boolean tryLock()</td> 
   <td>仅在调用时锁为空闲状态才获取该锁。如果锁可用，则获取锁，并立即返回值 true；如果锁不可用，则此方法将立即返回值 false。</td> 
  </tr> 
  <tr> 
   <td>boolean tryLock(long time, TimeUnit unit) throws InterruptedException</td> 
   <td>如果锁在给定的等待时间内空闲，并且当前线程未被中断，则获取锁。当前线程在以下3种情况下会返回：1.当前线程在超时时间内获得锁 2.当前线程在超时时间内被中断 3.已超过指定的等待时间</td> 
  </tr> 
  <tr> 
   <td>void unlock()</td> 
   <td>释放锁</td> 
  </tr> 
  <tr> 
   <td>Condition newCondition()</td> 
   <td>返回绑定到此 Lock 实例的新Condition实例。等待条件前，锁必须由当前线程保持，调用<code>Condition.await()</code>将在等待前以原子方式释放锁，并在等待返回前重新获取锁。</td> 
  </tr> 
 </tbody> 
</table>

**Lock接口提供的synchronized关键字所不具备主要特性：**

*  **尝试非阻塞的获取锁：** 当前线程尝试获取锁，如果这一时刻锁没有被其他线程获取到，则成功获取并持有锁；
 *  **能被中断的获取锁：** 与synchronized不同，获取到锁的线程能够响应中断，当获取到锁的线程被中断时，中断异常将会被抛出，同时锁会被释放；
 *  **超时获取锁：** 在指定的超时时间之前获取锁，如果截止时间到了仍旧无法获取锁，则返回。

Lock接口的实现基本都是通过聚合了一个同步器的子类来完成线程访问控制的，下面我们就来了解下这个同步器。

### 队列同步器AQS ###

队列同步器AbstractQueuedSynchronizer，是用来构建锁或其他同步组件的基础框架，它使用了一个int成员变量来表示同步状态，通过内置的FIFO队列来完成资源获取线程的排队。

同步器的主要使用方式是继承，子类通过继承同步器并实现它的抽象方法来管理同步状态。子类推荐被定义为自定义同步组件的静态内部类，同步器自身没有实现任何同步接口，它仅仅是定义了若干同步状态获取和释放的方法来供自定义同步组件使用。

可以这样理解同步器与锁的关系：锁是面向使用者的，它定义了使用者与锁交互的接口，隐藏了实现细节；同步器面向的是锁的实现者，它简化了锁的实现方式，屏蔽了同步状态管理、线程的排队、等待与唤醒等底层操作。

#### 同步器中的接口 ####

同步器的设计是基于模板方法模式的，使用者需要继承同步器并重写指定的方法，随后将同步器组合在自定义同步组件的实现中，并调用同步器提供的模板方法，而这些模板方法会调用使用者重写的方法。

**同步器提供的基础方法：用来访问和修改同步状态**

*  `getState()`：获取当前同步状态；
 *  `setState(int newState)`：设置当前同步状态；
 *  `compareAndSetState(int expect, int update)`：使用CAS设置当前同步状态，该方法能够保证状态设置的原子性。

**同步器可重写的方法**

*  `tryAcquire(int arg)`：独占式获取同步状态，实现该方法需要查询当前状态并判断同步状态是否符合预期，然后再进行CAS设置同步状态；
 *  `tryRelease(int arg)`：独占式释放同步状态；
 *  `tryAcquireShared(int arg)`：共享式获取同步状态，返回值大于等于0表示获取成功，反之获取失败；
 *  `tryReleaseShared(int arg)`：共享式释放同步状态；
 *  `isHeldExclusively()`：如果对于当前（正调用的）线程，同步是以独占方式进行的，则返回 true。

**同步器提供的模板方法**

*  `acquire(int arg)`：独占式获取同步状态，如果当前线程获取同步状态成功，则由该方法返回，否则进入同步队列等待，该方法将会调用重写的`tryAcquire(int arg)`方法；
 *  `acquireInterruptibly(int arg)`：与acquire(int arg)相同，但是该方法相应中断，如果被中断，则该方法会抛出InterruptedException返回；
 *  `tryAcquireNanos(int arg, long nanos)`：在acquireInterruptibly(int arg)基础上增加了超时限制，如果到了给定超时时间则返回fasle；
 *  `acquireShared(int arg)`：共享式获取同步状态，与独占式获取的主要区别是在同一时刻可以有多个线程获取到同步状态；
 *  `acquireSharedInteeruptibly(int arg)`：与acquireShared(int arg)相同，该方法响应中断；
 *  `tryAcquireSharedNanos(int arg, long nanos)`：在acquireSharedInteeruptibly(int arg)基础上增加了超时限制；
 *  `release(int arg)`：独占式释放同步状态；
 *  `releaseShared(int arg)`：共享式释放同步状态；
 *  `getQueuedThreads()`：获取等待在同步队列上的线程集合。

**下面利用AQS同步器来实现一个自定义锁：**

public class MyLock implements Lock {
        private final AQSHelper aqsHelper = new AQSHelper();
    
        @Override
        public void lock() {
            aqsHelper.acquire(1);
        }
    
        @Override
        public void lockInterruptibly() throws InterruptedException {
            aqsHelper.acquireInterruptibly(1);
        }
    
        @Override
        public boolean tryLock() {
            return aqsHelper.tryAcquire(1);
        }
    
        @Override
        public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
            return aqsHelper.tryAcquireNanos(1, unit.toNanos(time));
        }
    
        @Override
        public void unlock() {
            aqsHelper.release(1);
        }
    
        @Override
        public Condition newCondition() {
            return null;
        }
    
        /**
         * 队列同步器
         */
        private static class AQSHelper extends AbstractQueuedSynchronizer {
            // 是否处于独占状态
            @Override
            protected boolean isHeldExclusively() {
                return getState() == 1;
            }
    
            // 获取锁
            @Override
            protected boolean tryAcquire(int arg) {
                if (compareAndSetState(0, arg)) {
                    setExclusiveOwnerThread(Thread.currentThread());
                    return true;
                }
                return false;
            }
    
            // 释放锁
            @Override
            protected boolean tryRelease(int arg) {
                if (getState() == 0)
                    throw new IllegalMonitorStateException();
                setExclusiveOwnerThread(null);
                setState(0);
                return true;
            }
    
            // 返回一个Condition
            Condition newCondition() {
                return new ConditionObject();
            }
        }
    }

AQSHelper实现了独占式获取和释放同步状态，它被定义为同步组件MyLock的一个静态内部类。用户使用MyLock时并不会直接和内部的AQSHelper打交道，而是调用MyLock的方法。

**我们编写一个测试类，来测试下上面自定义的MyLock锁：**

public class AQSDemo {
        MyLock myLock = new MyLock();
        private int m = 0;
    
        public int next() {
            try {
                // 随机休眠2s
                TimeUnit.SECONDS.sleep(new Random().nextInt(2));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            myLock.lock();
            try {
                return m++;
            } finally {
                myLock.unlock();
            }
        }
    
        public static void main(String[] args) {
            AQSDemo aqsDemo = new AQSDemo();
            Thread[] th = new Thread[20];
            for (int i = 0; i < 20; i++) {
                th[i] = new Thread(() -> {
                    System.out.println(aqsDemo.next());
                });
                th[i].start();
            }
        }
    }

不加锁会出现重号和调号的现象，加锁后线程正确同步，演示效果请自行尝试。

上面的MyLock虽然解决了线程同步的问题，但是还有一个不足的地方，那就是不支持一个线程对资源的重复加锁，即重入性问题。

> 重入性是指，任意线程在获取到锁之后能够再次获取该锁而不会被锁所阻塞。

下面我们来改造下上面的代码，让其能够支持重入性。只需要修改AQSHelper中`tryAcquire()`和`tryRelease()`方法即可。

private class AQSHelper extends AbstractQueuedSynchronizer {
    		...
            // 获取锁
            @Override
            protected boolean tryAcquire(int arg) {
                int state = getState();
                if (state == 0) {
                    if (compareAndSetState(0, arg)) {
                        setExclusiveOwnerThread(Thread.currentThread());
                        return true;
                    }
                } else if (getExclusiveOwnerThread().equals(Thread.currentThread())) {
                    setState(getState() + arg);
                    return true;
                }
                return false;
            }
    
            // 释放锁
            @Override
            protected boolean tryRelease(int arg) {
                int state = getState() - arg;
                if (state == 0) {
                    setExclusiveOwnerThread(null);
                    setState(state);
                    return true;
                }
                setState(state);
                return false;
            }
            ...
        }

### Java并发包中提供的锁实现 ###

在实际开发中，除非是特殊需求，一般不需要自己去实现同步组件。`Java.util.concurrent`包里为我们提供了一些自带的同步组件实现，比如常用的ReentrantLock（可重入锁）、ReentrantReadWriteLock（可重入读写锁）、Semaphore、CountDownLatch、SynchronousQueue、FutureTask等，都是基于AQS构建的。想深入了解的同学可自行阅读源码，当你了解了AQS后，再去看这些源码会发现轻松很多。

下面简单介绍下锁的模式和类型  
**独占锁与共享锁**  
独占锁模式下，每次只能有一个线程能持有锁，其它获取锁的线程只能处于同步队列中等待，只有获取锁的线程释放了锁，后继的线程才能够获取锁，ReentrantLock就是以独占方式实现的排它锁；共享锁，则允许多个线程同时获取锁，并发访问共享资源，如：ReentrantReadWriteLock，读-写锁，它允许一个资源可以被多个读操作访问，或者被一个写操作访问，但两者不能同时进行。

独占锁是一种悲观保守的加锁策略，它避免了读/读冲突，如果某个只读线程获取锁，则其他读线程都只能等待，这种情况下就限制了不必要的并发性，因为读操作并不会影响数据的一致性。共享锁则是一种乐观锁，它放宽了加锁策略，允许多个执行读操作的线程同时访问共享资源。AQS的内部类Node定义了两个常量SHARED和EXCLUSIVE，他们分别标识 AQS队列中等待线程的锁获取模式。

static final class Node {
            /** Marker to indicate a node is waiting in shared mode */
            static final Node SHARED = new Node();
            /** Marker to indicate a node is waiting in exclusive mode */
            static final Node EXCLUSIVE = null;
    	...
    	}

**公平锁与非公平锁**  
公平锁是指线程将按他们发出请求的顺序来获得锁；而非公平锁则允许在线程发出请求后立即尝试获取锁，如果可用则可直接获取锁，尝试失败才进行排队等待。ReentrantLock和ReentrantReadWriteLock都支持公平性选择。

**最后来了解下ReentrantReadWriteLock，读写锁**  
ReadWriteLock接口表示存在两个锁：一个读取锁和一个写入锁，但是基于AQS实现的ReentrantReadWriteLock中，单个AQS子类将同时管理读取加锁和写入加锁ReentrantReadWriteLock使用了一个16位的状态来表示写入锁的计数，并且使用了另一个16位的状态来表示读取锁的计数。在读取锁上的操作将使用共享的获取与释放方法，在写入锁上的操作将使用独占的获取和释放方法。

ReentrantReadWriteLock有如下特性：

*  **公平性选择：** 支持非公平（默认）和公平的锁获取方式，吞吐量是非公平由于公平；
 *  **重进入：** 该锁支持重进入，以读写线程为例：读线程在获取锁后，能够再次获取读锁。而写线程在获取了写锁之后能够再次获取写锁，同时也可以获取读锁；
 *  **锁降级：** 遵循获取写锁、获取读锁再释放写锁的次序，写锁能够降级为读锁。

**代码示例**

public class Cache {
        static Map<String, Object> map = new HashMap<>();
        static ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
        static Lock r = rwl.readLock();
        static Lock w = rwl.writeLock();
    
        public static final Object get(String key) {
            r.lock();
            try {
                return map.get(key);
            } finally {
                r.unlock();
            }
        }
    
        public static final Object put(String key, Object value) {
            w.lock();
            try {
                return map.put(key, value);
            } finally {
                w.unlock();
            }
        }
    
        public static final void clear() {
            w.lock();
            try {
                map.clear();
            } finally {
                w.unlock();
            }
        }
    }

在上面的示例中，Cache组合了一个非线程安全的HashMap作为缓存的实现，同时使用读写锁的读锁和写锁来保证Cache是线程安全的。Cache使用读写锁提升读操作的并发性，也保证每次写操作对所有读写操作的可见性，同时简化了编程方式。

**参考资料**  
Java并发编程的艺术 方腾飞 魏鹏 程晓明 著

--------------------

\------------本文结束感谢您的阅读------------