Java并发编程之Lock

Posted 2020-10-31 第七狙击手

重入锁ReentrantLock

可以代替synchronized, 但synchronized更灵活.
但是, 必须必须必须要手动释放锁.

try {
    lock.lock();
} finally {
    lock.unlock();
}

重入锁是指任意线程在获取到锁之后能够再次获取该锁而不会被阻塞.
对于ReentrantLock而言, 释放锁时, 锁定调用了n次lock()方法, 那么释放时就需要调用n次unlock()方法.

• tryLock()方法, tryLock(long timeout, TimeUnit unit)方法
  尝试锁定,
  此方法有返回值, 锁定返回true, 否则返回false.
  如果无法锁定或者在一定时间内无法锁定, 线程可以决定是否等待.
• lockInterruptibly()方法
  线程在请求锁定的时候被阻塞, 如果被interrupt, 则此线程会被唤醒并被要求处理InterruptedException.

再次获取同步状态逻辑:
通过判断当前线程是否为获取锁的线程来决定获取操作是否成功, 如果是获取锁的线程, 则将同步状态增加并返回true, 表示获取同步状态成功.
如果是公平锁, 则还需要判断同步队列中当前节点是否有前驱节点, 如果有, 则需要等待前驱线程获取锁并释放之后才能继续获取锁.

公平锁和非公平锁
公平与否是针对获取锁而言的, 公平的是指获取锁的顺序是按请求时间顺序的, 也就是FIFO.
ReentrantLock默认为非公平锁, 构造对象时调用有参构造方法并参入true, 即为公平锁.

读写锁ReentrantReadWriteLock

读写锁在同一时刻允许多个读线程访问, 但是在写线程访问时, 所有的读线程和写线程都会被阻塞.
读写锁维护了一个读锁, 一个写锁. 并且, 遵循获取写锁再获取读锁, 再释放写锁的次序, 写锁可以阶级为读锁(即锁降级)

ReentrantReadWriteLock也支持重入和公平性选择

ReentrantReadWriteLock实现了ReadWriteLock接口, 此接口中只定义了获取读锁和写锁的方法, 即readLock()和writeLock()两个方法. ReentrantReadWriteLock还提供了如下方法:

方法名称	描述
int getReadLockCount()	返回当前读锁被获取的次数. 该次数不等于获取读锁的线程数, 一个线程可多次获取读锁
int getReadHoldCount()	返回当前线程获取读锁的次数, 并使用ThreadLocal保存
boolean isWriteLocked()	判断写锁是否被获取
int getWriteHoldCount()	返回当前写锁被获取的次数

ReentrantReadWriteLock通过将AQS的同步状态, 分为高16位(读)和低16位(写)来维护读写锁的获取状态,

图中状态表示, 当前线程已经获取了写锁, 并且重入了两次, 同时获取了两次读锁.
读写锁是通过位运算来确定读写状态的.
设当前同步状态为S, 那么:
写状态为 S & 0x0000FFFF, 即把高位16位(读)抹去
读状态为 S >>> 16, 即右移16位
写状态增加1时, S + 1
读状态增加1时, S + (1 << 16), 即 S + 0x00010000
推论: 同步状态S不等于0时, 当写状态(S & 0x0000FFFF)等于0时, 则读状态(S >>> 16)大于0, 即读锁已被获取.

写锁的获取与释放

写锁支持重入, 但是它是排它锁.
当前线程在获取写锁时:

1. 如果当前线程已经获取了写锁, 则写状态增加
2. 如果读锁已经被获取或者该线程不是已经获取写锁的线程, 则当前线程进行等待状态

protected final boolean tryAcquire(int acquires) {
    /*
     * Walkthrough:
     * 1. if read count nonzero or write count nonzero
     *     and owner is a different thread, fail.
     * 2. If count would saturate, fail. (This can only
     *    happen if count is already nonzero.)
     * 3. Otherwise, this thread is eligible for lock if
     *    it is either a reentrant acquire or
     *    queue policy allows it. If so, update state
     *    and set owner.
     */
    Thread current = Thread.currentThread();
    int c = getState();
    int w = exclusiveCount(c); // 获取写状态
    if (c != 0) {
        // (Note: if c != 0 and w == 0 then shared count != 0), 即上面的推论
        if (w == 0 || current != getExclusiveOwnerThread()) // 后面是重入条件
            return false;
        if (w + exclusiveCount(acquires) > MAX_COUNT)
            throw new Error("Maximum lock count exceeded");
    }
    if ((w == 0 && writerShouldBlock(current)) ||
        !compareAndSetState(c, c + acquires))
        return false;
    setExclusiveOwnerThread(current);
    return true;
}

如果存在读锁, 写锁不能被获取, 因为:
读写锁要确保写锁的操作对读锁可见, 如果允许读锁在已经被获取的情况下对写锁的获取, 那么正在运行的读线程就无法感知到写线程的操作.

写锁的释放, 就是每次释放写状态减小, 写状态为0时表示写锁已释放.

读锁的获取与释放

读锁是支持重入的共享锁.
读锁可以同时被多个线程获取, 在没有写线程访问的情况下, 读锁总是能被成功获取(读状态加增加).

Condition接口

在Java5之前, 等待/通知模式的实现可以采用wait()/notify()/notifyAll()与synchronized配合来实现.

任意Java对象上都有上述方法, 称为监视器方法

Condition接口也提供了类似的方法:

方法名称	描述
await()	当前线程进入等待状态, 直到被通知或中断
awaitUninterruptibly()	当前线程进入等待状态, 直到被通知
awaitNanos(long nanosTimeout)	当前线程进入等待状态, 直到被通知或中断或超时, 返回值为剩余时间
awaitUtil(Date deadLine)	当前线程进入等待状态, 直到被通知或中断或到达某个时间, 没到某个时间被通知返回true
signal()	唤醒一个等待在Condition上的线程, 该线程从等待方法返回前必须获得与Condition相关的锁
signalAll()	唤醒所有等待在Condition上的线程, 可以从等待方法返回的线程必须获得与Condition相关的锁

获取Condition对象必须通过Lock的newCondition()方法.

Lock lock = new ReentrantLock();
Condition con1 = lock.newCondition();
Condition con2 = lock.newCondition();

使用时必须先获取锁, 再调用condition的方法, 下面是使用Condition实现一个生产者/消费者场景:

public class MyContainer2<T> {
	final private LinkedList<T> lists = new LinkedList<>();
	final private int MAX = 10; //最多10个元素
	private int count = 0;
	
	private Lock lock = new ReentrantLock();
	private Condition producer = lock.newCondition();
	private Condition consumer = lock.newCondition();
	
	public void put(T t) {
		try {
			lock.lock();
			while(lists.size() == MAX) { //想想为什么用while而不是用if？
				producer.await();
			}
			
			lists.add(t);
			++count;
			consumer.signalAll(); //通知消费者线程进行消费
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
			lock.unlock();
		}
	}
	
	public T get() {
		T t = null;
		try {
			lock.lock();
			while(lists.size() == 0) {
				consumer.await();
			}
			t = lists.removeFirst();
			count --;
			producer.signalAll(); //通知生产者进行生产
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
			lock.unlock();
		}
		return t;
	}
	
	public static void main(String[] args) {
		MyContainer2<String> c = new MyContainer2<>();
		//启动消费者线程
		for(int i=0; i<10; i++) {
			new Thread(()->{
				for(int j=0; j<5; j++) System.out.println(c.get());
			}, "c" + i).start();
		}
		
		try {
			TimeUnit.SECONDS.sleep(2);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		
		//启动生产者线程
		for(int i=0; i<2; i++) {
			new Thread(()->{
				for(int j=0; j<25; j++) c.put(Thread.currentThread().getName() + " " + j);
			}, "p" + i).start();
		}
	}
}

参考资料: 《Java并发编程的艺术》