Java并发编程(三)

Java并发编程(三) - Lock详解

1. 概述

---

这里主要讲Java并发包中Lock的实现机制。

2. 队列同步器AQS(AbstractQueuedSynchronizer)

---

注：AQS主要是锁实现的原理，所以仅仅需要知道原理即可。这里的内容主要来自《Java并发编程的艺术》，然后AQS的内置FIFO队列我没有太看明白，所以没有写。
2.1 AQS的概述
队列同步器AQS(AbstractQueuedSynchronizer)是用来构件锁或者其他同步组件的基础框架，它使用一个volatile int成员变量来表示同步状态(其通过volatile特性与CAS特性来实现原子性以及内存可见性)，通过内置的FIFO队列完成资源获取线程的排队工作。

队列同步器AQS既支持独占访问，也支持共享式访问。同步组件，如：ReentrantLock、ReentrantReadWriteLock和CountDownLatch等等都是依赖于AQS来实现的。注：独占模式下指，其他线程试图获取该锁将无法取得成功。共享模式指，多个线程获取某个锁可能（但不是一定）会获得成功。

当一个线程去获得AQS的状态值时，如果获得成功就继续执行，否则AQS会将该线程放置同步队列中并阻塞该线程。AQS中可以有几个等待队列(通过Condition来获得)，AQS通过等待/通知机制来实现线程间的通知。其中阻塞、唤醒线程通过LockSupport中park()、unPark()等方法调用本地方法(JNI)来实现的，CAS的原子性以及volatile的内存可见性都是通过调用本地方法来实现的。

2.2 AQS的应用
AQS是基于模板方法进行设计的，使用者继承AQS并重写某些方法，然后调用AQS提供的模板方法，而这些模板方法将会调用使用者重写的方法实现功能。

AQS提供3个方法来访问或修改同步状态：

getState()：获取当前状态
setState(int newState)：设置同步状态
compareAndSetState(int expect, int update)：使用CAS方式来设置当前状态，该方法能够保证状态设置的原子性以及内存可见性

使用者继承AQS可重写的方法：

方法名称	方法描述
protected boolean tryAcquire(int arg)	试图在独占模式下获取对象状态，实现该方法需要查询当前状态并判断同步状态是否符合预期，然后在进行CAS设置同步状态
protected boolean tryRelease(int arg)	试图在独占模式下释放同步状态，使得同步队列中的线程可以获得同步状态
protected int tryAcquireShared(int arg)	试图共享式获取同步状态，返回值大于等于0，表示成功，反之失败
protected boolean tryReleaseShared(int arg)	试图共享式的释放同步状态
protected boolean isHeldExclusively()	当前AQS释放在独占模式下被线程锁占用

AQS通过给使用者的部分模板方法：

方法	方法描述
public final void acquire(int arg)	以独占模式获取对象，忽略中断
public final void acquireInterruptibly(int arg)	以独占模式获取对象，如果被中断则中止
public final boolean release(int arg)	以独占模式释放对象
public final boolean releaseShared(int arg)	以共享模式释放对象

下面是通过继承的方式实现互斥锁(Java API上面介绍AQS时使用)

public class Mutex implements Lock {
    private static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = 2178794844605968697L;

        @Override
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }

        @Override
        protected boolean tryAcquire(int arg) {
            if (compareAndSetState(0, 1)) {
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        @Override
        protected boolean tryRelease(int arg) {
            assert arg == 1; // Otherwise unused
            if (getState() == 0) throw new IllegalMonitorStateException();
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        Condition newCondition() {
            return new ConditionObject();
        }
    }

    private final 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 false;
    }

    @Override
    public void unlock() {
        sync.release(1);
    }

    @Override
    public Condition newCondition() {
        return sync.newCondition();
    }
}

3. Lock

---

3.1 Lock的写法
使用Lock的基本方式如下：

Lock lock = new ReentrantLock();
lock.lock();
try {
    处理逻辑
} finally {
    lock.unlock();
}

3.2 Lock接口提供的synchroinzed关键字不具备的特性

特性说明	API
尝试非阻塞的获取锁	boolean tryLock()
能被中断的获取锁	void lockInterruptibly()
超时获取锁	boolean tryLock(long time, TimeUtil unit)

4. 重入锁ReentrantLock

---

4.1 ReentrantLock的定义
重入锁(ReentrantLock)是支持重进入的锁，它表示该锁可以容许一个线程多次对资源重复加锁。该锁还支持获取锁时的公平和非公平性选择，注：synchronized关键字隐式支持重进入。

4.2 重进入的实现
下面是ReentrantLock的nonfairTryAcquire()、tryRelease()方法的实现：

final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        if (compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    } else if (current == getExclusiveOwnerThread()) {
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
         setState(nextc);
         return true;
    }
    return false;
}
protected final boolean tryRelease(int releases) {
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}

从上面代码我们可以看出来，首先要判断是否是当前线程，然后AQS的状态值表示该线程重入的次数。

4.3 公平锁和非公平锁
公平锁与非公平锁通过向ReentrantLock的构造函数中传递参数来获取。公平锁与非公平锁的主要区别在同步队列中的线程是否按照进入时间顺序排序。

5. 读写锁ReentrantReadWriteLock

---

5.1 ReentrantReadWriteLock的定义
之前的锁(ReentrantLock或者synchronized)都是排它锁，这些锁在同一时刻只能容许一个线程进行访问，而读写锁能够容许同一时刻多个读线程同时访问，但在写线程访问时，所有的读线程和其他写线程都被阻塞。

读写锁通过维护一个读锁一个写锁，通过分离读锁和写锁，使得并发性相比一般的排它锁有了很大的提高。

读写锁的实例如下，实例就像HashMap加锁一样，但是比Collections.synchronizedMap()性能要好，因为它使用的是读写锁。

public class Cache<K, V> {
    private Map<K, V> map = new HashMap<>();
    private ReentrantReadWriteLock readWriteLock = new ReentrantReadWriteLock();
    private Lock readLock = readWriteLock.readLock();
    private Lock writeLock = readWriteLock.writeLock();
    
    public V get(K key) {
        readLock.lock();
        try {
            return map.get(key);
        } finally {
            readLock.unlock();
        }
    }
    
    public V put(K key, V value) {
        writeLock.lock();
        try {
            return map.put(key, value);
        } finally {
            writeLock.unlock();
        }
    }
}

6. Condition

---

Condition接口
Condition顾名思义是条件的意思。它与Lock配合起来可以实现等待/通知机制，阻塞队列就是这样实现的，后面会写。前面已经写过了Condition+Lock的用法。

Condition+Lock与synchronized+Object的区别
Condition+Lock是基于AQS的，它的等待/通知机制与synchronized+Object的原理基本一致，有等待队列+同步队列构成。只是其等待队列可以有多个(一个Condition有一个等待队列)，这样就在多个线程通信时不用多个对象的等待+通知了。详情可以Java并发编程(二)里面关于消费者/生产者的例子

7. 应用

---

使用synchronized+object模拟的阻塞队列

public class ArrayBlockQueue1<E> {
    private Object[] elements;
    private final Object notFull = new Object();
    private final Object notEmpty = new Object();
    /** items index for next take, poll, peek or remove */
    private int takeIndex;
    /** items index for next put, offer, or add */
    private int putIndex;
    /** Number of elements in the queue */
    private int count;

    ArrayBlockQueue1() {
        this(10);
    }

    ArrayBlockQueue1(int initialSize) {
        elements = new Object[initialSize];
    }

    public void put(E e) throws InterruptedException {
        synchronized (notFull) {
            while (count == elements.length) {
                notFull.wait();
            }
        }
        synchronized (notEmpty) {
            final Object[] elements = this.elements;
            elements[putIndex] = e;
            if (++putIndex == elements.length) {
                putIndex = 0;
            }
            count++;
            notEmpty.notify();
        }
    }

    @SuppressWarnings("unchecked")
    public E take() throws InterruptedException {
        synchronized (notEmpty) {
            while (count == 0) {
                notEmpty.wait();
            }
        }
        synchronized (notFull) {
            final Object[] elements = this.elements;
            E e = (E) elements[takeIndex];
            elements[takeIndex] = null;
            if (++takeIndex == elements.length) {
                takeIndex = 0;
            }
            count--;
            notFull.notify();
            return e;
        }
    }
}

使用的Lock+Condition模拟的阻塞队列

public class ArrayBlockQueue<E> {
    private Object[] elements;
    private ReentrantLock lock = new ReentrantLock();
    private Condition notFull = lock.newCondition();
    private Condition notEmpty = lock.newCondition();
    /** items index for next take, poll, peek or remove */
    private int takeIndex;
    /** items index for next put, offer, or add */
    private int putIndex;
    /** Number of elements in the queue */
    private int count;

    ArrayBlockQueue() {
        this(10);
    }

    ArrayBlockQueue(int initialSize) {
        elements = new Object[initialSize];
    }

    public void put(E e) throws InterruptedException {
        if (e == null) {
            throw new NullPointerException();
        }
        lock.lock();
        try {
            while (count == elements.length) {
                notFull.await();
            }
            addQueue(e);
        } finally {
            lock.unlock();
        }
    }

    private void addQueue(E e) {
        final Object[] elements = this.elements;
        elements[putIndex] = e;
        if (++putIndex == elements.length) {
            putIndex = 0;
        }
        count++;
        notEmpty.signal();
    }

    public E take() throws InterruptedException {
        lock.lock();
        try {
            while (count == 0) {
                notEmpty.await();
            }
            return removeQueue();
        } finally {
            lock.unlock();
        }
    }

    @SuppressWarnings("unchecked")
    private E removeQueue() {
        final Object[] elements = this.elements;
        E element = (E) elements[takeIndex];
        elements[takeIndex] = null;
        if (++takeIndex == elements.length)
            takeIndex = 0;
        count--;
        notFull.signal();
        return element;
    }
}

8.References

---

《Java并发编程的艺术》