LockSupport

concurrent包是基于AQS (AbstractQueuedSynchronizer)框架的,AQS框架借助于两个类:

  • Unsafe(提供CAS操作)
  • LockSupport(提供park/unpark操作)

因此,LockSupport非常重要。
两个重点
(1)操作对象
归根结底,LockSupport.park()和LockSupport.unpark(Thread thread)调用的是Unsafe中的native代码:

//LockSupport中
public static void park() {
     UNSAFE.park(false, 0L);
}
//LockSupport中
public static void unpark(Thread thread) {
        if (thread != null)
            UNSAFE.unpark(thread);
    }

Unsafe类中的对应方法:

    //park
    public native void park(boolean isAbsolute, long time);
    
    //unpack
    public native void unpark(Object var1);

park函数是将当前调用Thread阻塞,而unpark函数则是将指定线程Thread唤醒。

与Object类的wait/notify机制相比,park/unpark有两个优点:
以thread为操作对象更符合阻塞线程的直观定义
操作更精准,可以准确地唤醒某一个线程(notify随机唤醒一个线程,notifyAll唤醒所有等待的线程),增加了灵活性。

(2)关于“许可”

在上面的文字中,我使用了阻塞和唤醒,是为了和wait/notify做对比。

其实park/unpark的设计原理核心是“许可”:park是等待一个许可,unpark是为某线程提供一个许可。
如果某线程A调用park,那么除非另外一个线程调用unpark(A)给A一个许可,否则线程A将阻塞在park操作上。

有一点比较难理解的,是unpark操作可以再park操作之前。
也就是说,先提供许可。当某线程调用park时,已经有许可了,它就消费这个许可,然后可以继续运行。这其实是必须的。考虑最简单的生产者(Producer)消费者(Consumer)模型:Consumer需要消费一个资源,于是调用park操作等待;Producer则生产资源,然后调用unpark给予Consumer使用的许可。非常有可能的一种情况是,Producer先生产,这时候Consumer可能还没有构造好(比如线程还没启动,或者还没切换到该线程)。那么等Consumer准备好要消费时,显然这时候资源已经生产好了,可以直接用,那么park操作当然可以直接运行下去。如果没有这个语义,那将非常难以操作。

但是这个“许可”是不能叠加的,“许可”是一次性的。
比如线程B连续调用了三次unpark函数,当线程A调用park函数就使用掉这个“许可”,如果线程A再次调用park,则进入等待状态。

Unsafe.park和Unsafe.unpark的底层实现原理

在Linux系统下,是用的Posix线程库pthread中的mutex(互斥量),condition(条件变量)来实现的。
mutex和condition保护了一个_counter的变量,当park时,这个变量被设置为0,当unpark时,这个变量被设置为1。

源码:
每个Java线程都有一个Parker实例,Parker类是这样定义的:

class Parker : public os::PlatformParker {  
private:  
  volatile int _counter ;  
  ...  
public:  
  void park(bool isAbsolute, jlong time);  
  void unpark();  
  ...  
}  
class PlatformParker : public CHeapObj<mtInternal> {  
  protected:  
    pthread_mutex_t _mutex [1] ;  
    pthread_cond_t  _cond  [1] ;  
    ...  
}  

可以看到Parker类实际上用Posix的mutex,condition来实现的。
在Parker类里的_counter字段,就是用来记录“许可”的。

park 过程

当调用park时,先尝试能否直接拿到“许可”,即_counter>0时,如果成功,则把_counter设置为0,并返回:

void Parker::park(bool isAbsolute, jlong time) {  
  
  // Ideally we'd do something useful while spinning, such  
  // as calling unpackTime().  
  
  // Optional fast-path check:  
  // Return immediately if a permit is available.  
  // We depend on Atomic::xchg() having full barrier semantics  
  // since we are doing a lock-free update to _counter.  
  
  if (Atomic::xchg(0, &_counter) > 0) return;  

如果不成功,则构造一个ThreadBlockInVM,然后检查_counter是不是>0,如果是,则把_counter设置为0,unlock mutex并返回:

ThreadBlockInVM tbivm(jt);  
if (_counter > 0)  { // no wait needed  
  _counter = 0;  
  status = pthread_mutex_unlock(_mutex);  

否则,再判断等待的时间,然后再调用pthread_cond_wait函数等待,如果等待返回,则把_counter设置为0,unlock mutex并返回:

if (time == 0) {  
  status = pthread_cond_wait (_cond, _mutex) ;  
}  
_counter = 0 ;  
status = pthread_mutex_unlock(_mutex) ;  
assert_status(status == 0, status, "invariant") ;  
OrderAccess::fence();  

unpark 过程
当unpark时,则简单多了,直接设置_counter为1,再unlock mutex返回。如果_counter之前的值是0,则还要调用pthread_cond_signal唤醒在park中等待的线程:

void Parker::unpark() {  
  int s, status ;  
  status = pthread_mutex_lock(_mutex);  
  assert (status == 0, "invariant") ;  
  s = _counter;  
  _counter = 1;  
  if (s < 1) {  
     if (WorkAroundNPTLTimedWaitHang) {  
        status = pthread_cond_signal (_cond) ;  
        assert (status == 0, "invariant") ;  
        status = pthread_mutex_unlock(_mutex);  
        assert (status == 0, "invariant") ;  
     } else {  
        status = pthread_mutex_unlock(_mutex);  
        assert (status == 0, "invariant") ;  
        status = pthread_cond_signal (_cond) ;  
        assert (status == 0, "invariant") ;  
     }  
  } else {  
    pthread_mutex_unlock(_mutex);  
    assert (status == 0, "invariant") ;  
  }  
}  

JDK1.8后,ReentrantLock及ReentrantReadWriteLock是基于AQS实现的,AQS内部使用了unsafe类进行操作;
LockSupport也是基于unsafe类操作。
可以说LockSupport也是阻塞的,但是不会发生Thread.suspend 和 Thread.resume所可能引发的死锁问题。
而AQS是非阻塞机制。
LockSupport.park()和unpark(),与object.wait()和notify()的区别?
主要的区别应该说是它们面向的对象不同。阻塞和唤醒是对于线程来说的,LockSupport的park/unpark更符合这个语义,以“线程”作为方法的参数, 语义更清晰,使用起来也更方便。而wait/notify的实现使得“阻塞/唤醒对线程本身来说是被动的,要准确的控制哪个线程、什么时候阻塞/唤醒很困难, 要不随机唤醒一个线程(notify)要不唤醒所有的(notifyAll)。

代码示例:

package juc;

import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.LockSupport;

public class TestLockSupport {

	static final ExecutorService executorService = Executors.newCachedThreadPool();

	public static void main(String[] args) {
		FIFOMutex mutex = new FIFOMutex();
		for (int i = 0; i < 10; i++) {
			executorService.submit(new Runnable() {

				@Override
				public void run() {
					mutex.lock();
					String threadName = Thread.currentThread().getName();
					System.out.println(threadName + " : start!");
					try {
						Thread.sleep(2000);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					System.out.println(threadName + " : done!");
					mutex.unlock();
				}
			});
		}

	}

}

class FIFOMutex {
	private final AtomicBoolean locked = new AtomicBoolean(false);
	private final Queue<Thread> waiters = new ConcurrentLinkedQueue<Thread>();

	public void lock() {
		boolean wasInterrupted = false;
		Thread current = Thread.currentThread();
		waiters.add(current);

		// Block while not first in queue or cannot acquire lock
		// 如果不是第一个 或者 没有获取到锁就阻塞
		while (waiters.peek() != current || !locked.compareAndSet(false, true)) {
			LockSupport.park(this);
			// 等待过程忽略中断
			if (Thread.interrupted()) // ignore interrupts while waiting
				wasInterrupted = true;
		}

		waiters.remove();
		// 恢复中断状态
		if (wasInterrupted) // reassert interrupt status on exit
			current.interrupt();
	}

	public void unlock() {
		locked.set(false);
		LockSupport.unpark(waiters.peek());
	}
}

作者:SinX竟然被占用了
链接:https://www.jianshu.com/p/e3afe8ab8364
来源:简书
著作权归作者所有。商业转载请联系作者获得授权,非商业转载请注明出处。