高并发编程之LockSupport深入源码剖析_LoveIT的博客

与Object的wait()/notify()相比：

wait()、notify()、notifyAll()都必须配置Object monitor使用，而LockSupport的park()、unpark()方法不必
park()、unpark()可以精确到一个线程来阻塞和唤醒它，但是notify()只能在Wait Set中随机唤醒一个，并且notifyAll()也只能是唤醒所有的线程，控制精度不及unpark()。
一个线程执行unpark()之前可以调用park()，并且线程执行的效果和先调用park()再调用unpark()一致。

1、LockSupport的基本使用

LockSupport中有一组方法可以实现对某一个线程精确的阻塞和唤醒，即：park()及他的重载方法和unpark()

（1）正常使用：先阻塞再唤醒

package top.easyblog;

import java.util.concurrent.locks.LockSupport;

/**
 * @author ：huangxin
 * @modified ：
 * @since ：2020/04/01 17:06
 */
public class LockSupportTest {

    public static void main(String[] args) {
        Thread t1 = new Thread(() -> {
            System.out.println("start");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("park.....");
            LockSupport.park();
            System.out.println("resume.....");
        },"t1");
        t1.start();
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("unpark...");
        LockSupport.unpark(t1);
    }
}

执行结果：

（2）先唤醒再阻塞

package top.easyblog;

import java.util.concurrent.locks.LockSupport;

/**
 * @author ：huangxin
 * @modified ：
 * @since ：2020/04/01 17:06
 */
public class LockSupportTest {

    public static void main(String[] args) {
        Thread t1 = new Thread(() -> {
            System.out.println("start");
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("park.....");
            LockSupport.park();
            System.out.println("resume.....");
        },"t1");
        t1.start();
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("unpark...");
        LockSupport.unpark(t1);

    }
}

执行结果：

可以看到，这次是先执行了唤醒操作，之后执行了阻塞操作，但是程序并没有像我们预期的那样阻塞，而是正常执行结束了。这是为什么呢？接下来我们就深入到源码中探寻真相！！！

二、park()、unpark()原理剖析

在Java API层面很简单，最终都是调用了UnSafe类的对应的本地方法，为了探究底层原理，我找到了关于这个用C++实现的Parker类，**每一个线程都有一个与之关联的Parker对象，Parker对象主要由__counter、__cond和__mutex组成。**下面是Parker类的部分源码：

class Parker : public os::PlatformParker {  //Parker继承了PaltformaParker
private:
  volatile int _counter ;   //计数
  Parker * FreeNext ;      //指向下一个Parker
  JavaThread * AssociatedWith ;   // 指向parker所属的线程。
 
public:
  Parker() : PlatformParker() {
    _counter       = 0 ;    //初始化为0
    FreeNext       = NULL ;
    AssociatedWith = NULL ;
  }
protected:
  ~Parker() { ShouldNotReachHere(); }
public:
  // For simplicity of interface with Java, all forms of park (indefinite,
  // relative, and absolute) are multiplexed into one call.
  void park(bool isAbsolute, jlong time);
  void unpark();
 
  // Lifecycle operators
  static Parker * Allocate (JavaThread * t) ;
  static void Release (Parker * e) ;
private:
  static Parker * volatile FreeList ;
  static volatile int ListLock ;
 
};

//PlatformParker类
class PlatformParker : public CHeapObj<mtInternal> {
  protected:
    enum {
        REL_INDEX = 0,
        ABS_INDEX = 1
    };
    int _cur_index;    // 条件变量数组下标，which cond is in use: -1, 0, 1
    pthread_mutex_t _mutex [1] ;  //pthread互斥锁
    pthread_cond_t  _cond  [2] ; // pthread条件变量数组,一个用于相对时间，一个用于绝对时间。
 
  public:       // TODO-FIXME: make dtor private
    ~PlatformParker() { guarantee (0, "invariant") ; }
 
  public:
    PlatformParker() {
      int status;
      status = pthread_cond_init (&_cond[REL_INDEX], os::Linux::condAttr());
      assert_status(status == 0, status, "cond_init rel");
      status = pthread_cond_init (&_cond[ABS_INDEX], NULL);
      assert_status(status == 0, status, "cond_init abs");
      status = pthread_mutex_init (_mutex, NULL);
      assert_status(status == 0, status, "mutex_init");
      _cur_index = -1; // mark as unused
    }
};

LockSupport就是通过Parker类中的__counter实现对一个线程的阻塞和唤醒的。

调用park()示意图

Parker::park()：


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;

  Thread* thread = Thread::current();
  assert(thread->is_Java_thread(), "Must be JavaThread");
  JavaThread *jt = (JavaThread *)thread;

  // Optional optimization -- avoid state transitions if there's an interrupt pending.
  // Check interrupt before trying to wait
  if (Thread::is_interrupted(thread, false)) {
    return;
  }

  // Next, demultiplex/decode time arguments
  timespec absTime;
  if (time < 0 || (isAbsolute && time == 0) ) { // don't wait at all
    return;
  }
  if (time > 0) {
    unpackTime(&absTime, isAbsolute, time);
  }


  // Enter safepoint region
  // Beware of deadlocks such as 6317397.
  // The per-thread Parker:: mutex is a classic leaf-lock.
  // In particular a thread must never block on the Threads_lock while
  // holding the Parker:: mutex.  If safepoints are pending both the
  // the ThreadBlockInVM() CTOR and DTOR may grab Threads_lock.
  ThreadBlockInVM tbivm(jt);

  // Don't wait if cannot get lock since interference arises from
  // unblocking.  Also. check interrupt before trying wait
  if (Thread::is_interrupted(thread, false) || pthread_mutex_trylock(_mutex) != 0) {
    return;
  }

  int status ;
  if (_counter > 0)  { // no wait needed
    _counter = 0;
    status = pthread_mutex_unlock(_mutex);
    assert (status == 0, "invariant") ;
    // Paranoia to ensure our locked and lock-free paths interact
    // correctly with each other and Java-level accesses.
    OrderAccess::fence();
    return;
  }

#ifdef ASSERT
  // Don't catch signals while blocked; let the running threads have the signals.
  // (This allows a debugger to break into the running thread.)
  sigset_t oldsigs;
  sigset_t* allowdebug_blocked = os::Linux::allowdebug_blocked_signals();
  pthread_sigmask(SIG_BLOCK, allowdebug_blocked, &oldsigs);
#endif

  OSThreadWaitState osts(thread->osthread(), false /* not Object.wait() */);
  jt->set_suspend_equivalent();
  // cleared by handle_special_suspend_equivalent_condition() or java_suspend_self()

  assert(_cur_index == -1, "invariant");
  if (time == 0) {
    _cur_index = REL_INDEX; // arbitrary choice when not timed
    status = pthread_cond_wait (&_cond[_cur_index], _mutex) ;
  } else {
    _cur_index = isAbsolute ? ABS_INDEX : REL_INDEX;
    status = os::Linux::safe_cond_timedwait (&_cond[_cur_index], _mutex, &absTime) ;
    if (status != 0 && WorkAroundNPTLTimedWaitHang) {
      pthread_cond_destroy (&_cond[_cur_index]) ;
      pthread_cond_init    (&_cond[_cur_index], isAbsolute ? NULL : os::Linux::condAttr());
    }
  }
  _cur_index = -1;
  assert_status(status == 0 || status == EINTR ||
                status == ETIME || status == ETIMEDOUT,
                status, "cond_timedwait");

#ifdef ASSERT
  pthread_sigmask(SIG_SETMASK, &oldsigs, NULL);
#endif

  _counter = 0 ;
  status = pthread_mutex_unlock(_mutex) ;
  assert_status(status == 0, status, "invariant") ;
  // Paranoia to ensure our locked and lock-free paths interact
  // correctly with each other and Java-level accesses.
  OrderAccess::fence();

  // If externally suspended while waiting, re-suspend
  if (jt->handle_special_suspend_equivalent_condition()) {
    jt->java_suspend_self();
  }
}

park的流程如下：

step1.如果有许可可用，则将_counter原子地设置为0，并直接返回。 xchg返回的是旧的_counter；否则将没有许可可用。
step2.获取当前线程，如果当前线程设置了中断标志，则直接返回，因此如果在park前调用了interrupt就会直接返回。
step3.获取定时时间，安全点；如果中断或获取_mutex失败，则直接返回
step4.如果_counter=1，说明之前unpark已经调用过了。所以只需将_counter置为0，解锁返回（表现在程序上就是在执行LockSupport.park()后不会被阻塞）。
step5.对于time = 0，pthread_cond_wait (&_cond[_cur_index], _mutex) 直接挂起；对于定时的，挂起指定的时间status = os::Linux::safe_cond_timedwait (&_cond[_cur_index], _mutex, &absTime) ;

调用unpark()示意图

Parker::unpark()

void Parker::unpark() {
  int s, status ;
  status = pthread_mutex_lock(_mutex);
  assert (status == 0, "invariant") ;
  s = _counter;
  _counter = 1;
  if (s < 1) {
    // thread might be parked
    if (_cur_index != -1) {
      // thread is definitely parked
      if (WorkAroundNPTLTimedWaitHang) {
        status = pthread_cond_signal (&_cond[_cur_index]);
        assert (status == 0, "invariant");
        status = pthread_mutex_unlock(_mutex);
        assert (status == 0, "invariant");
      } else {
        // must capture correct index before unlocking
        int index = _cur_index;
        status = pthread_mutex_unlock(_mutex);
        assert (status == 0, "invariant");
        status = pthread_cond_signal (&_cond[index]);
        assert (status == 0, "invariant");
      }
    } else {
      pthread_mutex_unlock(_mutex);
      assert (status == 0, "invariant") ;
    }
  } else {
    pthread_mutex_unlock(_mutex);
    assert (status == 0, "invariant") ;
  }
}

unpark的运行流程：

step1.对_mutex加锁，并将_counter置为1。
step2.如果之前的_counter为0则说明调用了park或者为初始状态（此时为0且没有调用park）。_
- _step2-1.当前parker对应的线程挂起了。因为_cur_index初始化为-1，且线程唤醒后也会重置为-1。调用pthread_cond_signal (&_cond[_cur_index])。调用pthread_mutex_unlock(_mutex)
- step2-2.没有线程在等待条件变量，则直接解锁： pthread_mutex_unlock(_mutex);

step3.如果之前的_counter为1，则说明线程调用了一次或多次unpark但是没调用park，则直接解锁。

分析了这么多，总结就是：

在调用park的时候如果__counter是0则会去执行挂起的流程，否则返回，在挂起恢复后再将counter置为0。
在unpark的时候如果 __counter是0则会执行唤醒的流程，否则不执行唤醒流程，并且不管什么情况始终将 _counter重置为1。
在park里，调用pthread_cond_wait时，并没有用while来判断，所以posix condition里的"Spurious wakeup"一样会传递到上层Java的代码里（因为条件需要Java层才能提供）。这也就是为什么Java dos里提到需要注意虚假唤醒的情况。

高并发编程之LockSupport深入源码剖析

1、LockSupport的基本使用

二、park()、unpark()原理剖析

留言区