synchronized的锁优化原理

synchronized的用法

synchronized修饰的方法或代码块相当于并发中的临界区，即在同一时刻jvm只允许一个线程进入执行。synchronized是通过锁机制实现同一时刻只允许一个线程来访问共享资源的。另外synchronized锁机制还可以保证线程并发运行的原子性，有序性，可见性。

为了保证在同一时刻jvm只允许一个线程进入执行，jvm是如何来实现与优化的呢？
带着几个问题来看：
1.synchronized是如何保证线程互斥的？
2.对于同一个线程，又如何保证它可以重入？
3.为什么优化？如何优化？

在JVM中，线程安全的实现方法有三种：
1.互斥同步。

互斥同步，最常见的一种并发正确性保障手段。同步是指在多个线程并发访问共享数据时，在同一个时刻只有一个线程可以访问到共享数据。互斥是实现同步的一种手段，主要实现方式有临界区，信号量，互斥量。但是它的实现方式比较悲观，无论有没有多线程的竞争，都会对共享资源进行加锁。但是加锁的过程又是及其耗费资源的操作（用户态与内核态的切换）。
其实在很多时候是不存在竞争的情况，这样加锁也就没有太大必要，但是要保证在其他时候并发线程出现竞争，所以互斥同步又不得不加锁。我们称之为悲观锁。
同时，对于线程重入的情况，也没有必要再去加锁（用户态与内核态的切换），因为本身已经持有了，重复使用就可以了。
所以为了避免加锁切换状态带来的资源耗费，就出现了非阻塞同步方案。

2.非阻塞同步。

基于冲突检测的乐观并发策略，通俗来说就是不管资源有没有被锁，线程都会先进行操作：
a.如果没有其他线程使用共享资源，那么此线程就可以直接进行操作。
b.如果有其他线程使用共享资源，那么此线程发现了冲突，可以进行其他的补偿措施（比如不断重试，直到成功）。
这种同步方式不需要把线程挂起，也就被称为非阻塞同步。

3.无同步方案。

不涉及共享数据，就不需要同步。
比如可重入代码，线程本地储存。

synchronized最初就是使用了互斥同步来保证线程安全的。
在操作系统层面，互斥加锁是使用的mutex操作。但是由于操作的复杂性，有可能出现死锁的情况。而在高级语言中，提出了monitor机制，它把mutex的一些操作对上层屏蔽掉，在内部自己实现了这些机制，是我们可以更简单的使用同步。

monitor的基本元素

1.临界区
2.monitor对象及锁
3.条件变量以及在monitor对象上的wait，signal操作。

临界区就是被synchronized修饰的代码块/方法。
monitor对象就是synchronzied关联的对象，可以是任何的Object。Object的monitor机制是依赖于JVM中ObjectMonitor模式的实现，wait()，notify()，notifyAll()方法就是由这个实现的。关于monitor的原理

image.png

The Owner表示的是锁资源，进入到这里就表示获取到锁。
左边的entrySet表示被阻塞的线程，没有获取到锁资源时在这个容器里面等待。
右边的waitSet表示由于某些原因获取锁的线程不能向下进行下去，需要让出资源给其他线程，此时放到了waitSet中，等待外部条件允许的时候重新进入entrySet去竞争锁。这个外部条件在 monitor 机制中称为条件变量。

不过monitor只是屏蔽了对底层metux的实现，实际上还是需要去切换用户态/内核态来实现同步，monitor只是对实现的一种优化，并没有对性能进行优化。

jdk1.6后对于synchronized的优化，是通过偏向锁，轻量级锁，重量级锁的机制。
在某些情况下，线程的竞争没有很激烈，甚至没有竞争，使用互斥同步的方法得不偿失，因为内核态与用户态的切换时锁耗费的资源甚至比同步代码块的资源多得多。
轻量级锁就是使用cas操作消除同步使用的互斥量(metux).
偏向锁就是相当于把整个同步都消除掉。

关于锁的升级过程与原理，还需要从java的对象头中的MarkWord讲起。
在HotSpot虚拟机中，一个对象在内存储存的布局可以分为三个区域：对象头，实例数据，对齐填充。结构如下：

image.png

这里的重点是MarkWord，MarkWord中储存的内容结构是动态的，根据不同的锁状态会保存不同的数据。
无锁状态下，锁标记01，偏向锁标记0。
在升级为偏向锁时，偏向锁标记变成1，同时前面的数据中替换成偏向线程的id，用来记录偏向锁的偏向线程，目的是在一个线程进入到同步代码块中时，会检查当前线程与锁对象的偏向线程是否为同一个，如果是就会重入，否则就升级到轻量级锁。
升级到轻量级锁时，锁标记变为00，前面的内容变成了指向线程栈中锁记录的指针。轻量级锁的加锁过程如下：

当前线程栈帧中建立名为锁记录（Lock Record）的空间，用于存储MarkWord的拷贝（Displaced Mark Word）

image.png

然后虚拟机使用cas操作尝试讲对象的MarkWord更新为一个指向Lock Record的指针。如果成功代表获取锁成功，此时锁标记变为00。

image.png

如果更新操作失败，虚拟机会检查对象的MarkWord是否是指向当前线程的指针，是的话就说明被当前线程持有，可以重入。否则说明锁对象被其他线程占用，或者锁已经进行膨胀操作。
这里解释一下膨胀操作，并不是膨胀成重量级锁，在这里的操作中会有许多事情要做，来看下jvm的源码：
ObjectSynchronizer::slow_enter

void ObjectSynchronizer::slow_enter(Handle obj, BasicLock* lock, TRAPS) {
  markOop mark = obj->mark();
  assert(!mark->has_bias_pattern(), "should not see bias pattern here");

// 无锁状态下，直接设置线程栈中的Lock Record为markWord
  if (mark->is_neutral()) {
    // Anticipate successful CAS -- the ST of the displaced mark must
    // be visible <= the ST performed by the CAS.
    lock->set_displaced_header(mark);
    if (mark == (markOop) Atomic::cmpxchg_ptr(lock, obj()->mark_addr(), mark)) {
      TEVENT (slow_enter: release stacklock) ;
      return ;
    }
    // Fall through to inflate() ...
  } else // 有锁状态，并且发现是当前线程持有，即重入。重入的Lock Record的markWord设置为null
  if (mark->has_locker() && THREAD->is_lock_owned((address)mark->locker())) {
    assert(lock != mark->locker(), "must not re-lock the same lock");
    assert(lock != (BasicLock*)obj->mark(), "don't relock with same BasicLock");
    lock->set_displaced_header(NULL);
    return;
  }

#if 0
  // The following optimization isn't particularly useful.
  if (mark->has_monitor() && mark->monitor()->is_entered(THREAD)) {
    lock->set_displaced_header (NULL) ;
    return ;
  }
#endif

// 当发现有锁，又不是被当前线程持有，即有第二个线程来获取轻量级锁资源，那么进行膨胀操作。
// 膨胀操作只是返回一个ObjectMonitor对象，真正的
  // The object header will never be displaced to this lock,
  // so it does not matter what the value is, except that it
  // must be non-zero to avoid looking like a re-entrant lock,
  // and must not look locked either.
  lock->set_displaced_header(markOopDesc::unused_mark());
  ObjectSynchronizer::inflate(THREAD, obj())->enter(THREAD);
}

此时有第二个线程来获取锁资源的时候，就进行膨胀操作ObjectSynchronizer::inflate，返回一个ObjectMonitor
这个方法中，会根据当前的markWord的状态来进行不同的操作：

1.已经膨胀过(inflated)，直接获取markWord所指向的Monitor。
2.正在膨胀中（INFLATING），会直接继续循环重新判断状态。
3.被轻量级锁定（stack-locked），先分配一段空间给ObjectMonitor创建，再进行一些初始化操作。利用cas把markWord的状态改成膨胀中（INFLATING），此时如果失败，那么释放资源并重试。讲持有锁的线程栈中指向的markWord复制到ObjectMonitor的header，ObjectMonitor的owner设置为此时markword中指向的Lock Record，object设置为锁对象。也就是根据当前的状态新建一个ObjectMonitor，返回。
4.无锁状态(neutral)，也是新建一个ObjectMonitor，不过没有设置owner。并且把markWord的重量级锁互斥指针指向它。如果替换失败就进行下一次自旋。

得到了ObjectMonitor之后，使用其enter方法，

 cur = Atomic::cmpxchg_ptr (Self, &_owner, NULL) ;
// ObjectMonitor没有被占有，替换成功，结束。
  if (cur == NULL) {
     // Either ASSERT _recursions == 0 or explicitly set _recursions = 0.
     assert (_recursions == 0   , "invariant") ;
     assert (_owner      == Self, "invariant") ;
     // CONSIDER: set or assert OwnerIsThread == 1
     return ;
  }
// 发现是当前线程持有，重入标记+1
  if (cur == Self) {
     // TODO-FIXME: check for integer overflow!  BUGID 6557169.
     _recursions ++ ;
     return ;
  }
// 上面直到在轻量级锁膨胀的时候，会把owner设置为当时持有锁的线程栈上的Lock Record，这里就是表示为从轻量级锁第一次成功膨胀。
  if (Self->is_lock_owned ((address)cur)) {
    assert (_recursions == 0, "internal state error");
    _recursions = 1 ;
    // Commute owner from a thread-specific on-stack BasicLockObject address to
    // a full-fledged "Thread *".
    _owner = Self ;
    OwnerIsThread = 1 ;
    return ;
  }

接下来是替换失败的操作，

// 先在入队之前尝试自旋一圈，
  // Try one round of spinning *before* enqueueing Self
  // and before going through the awkward and expensive state
  // transitions.  The following spin is strictly optional ...
  // Note that if we acquire the monitor from an initial spin
  // we forgo posting JVMTI events and firing DTRACE probes.
  if (Knob_SpinEarly && TrySpin (Self) > 0) {
     assert (_owner == Self      , "invariant") ;
     assert (_recursions == 0    , "invariant") ;
     assert (((oop)(object()))->mark() == markOopDesc::encode(this), "invariant") ;
     Self->_Stalled = 0 ;
     return ;
  }

在TrySpin这个方法中，会根据条件来进行自旋获取锁

    for (ctr = Knob_PreSpin + 1; --ctr >= 0 ; ) {
      if (TryLock(Self) > 0) {
        // Increase _SpinDuration ...
        // Note that we don't clamp SpinDuration precisely at SpinLimit.
        // Raising _SpurDuration to the poverty line is key.
        int x = _SpinDuration ;
        if (x < Knob_SpinLimit) {
           if (x < Knob_Poverty) x = Knob_Poverty ;
           _SpinDuration = x + Knob_BonusB ;
        }
        return 1 ;
      }
      SpinPause () ;
    }

看看TryLock

int ObjectMonitor::TryLock (Thread * Self) {
   for (;;) {
      void * own = _owner ;
      if (own != NULL) return 0 ;
      if (Atomic::cmpxchg_ptr (Self, &_owner, NULL) == NULL) {
         // Either guarantee _recursions == 0 or set _recursions = 0.
         assert (_recursions == 0, "invariant") ;
         assert (_owner == Self, "invariant") ;
         // CONSIDER: set or assert that OwnerIsThread == 1
         return 1 ;
      }
      // The lock had been free momentarily, but we lost the race to the lock.
      // Interference -- the CAS failed.
      // We can either return -1 or retry.
      // Retry doesn't make as much sense because the lock was just acquired.
      if (true) return -1 ;
   }
}

也就是说，当ObjectMonitor中owner有值时，表示被占用，继续自旋。
没有值的时候，并且被替换为当前线程时候，表示获取锁成功，停止自旋。
在真正的入队之前，会通过自旋来优化性能。达到一定次数之后，就会入队了，交接给系统的metux。

到这里我们的锁升级过程就清晰了，流程如图

image.png

总结一下三种锁的优缺点

锁类型	优点	缺点	适用场景
偏向锁	没有加锁解锁的额外开销，相当于消除同步代码块，性能影响最小	如果存在竞争时，在线程撤销锁时会存在额外的开销	只有一个线程访问同步代码块的场景
轻量级锁	线程竞争时，不存在排队等待，不需要内核态与用户态的切换，开销比较低	大量线程竞争时，自选操作浪费了cpu资源	同步块执行很快，追求响应时间
重量级锁	线程竞争不使用自旋，不会消耗CPU。	线程阻塞，响应时间缓慢。	追求吞吐量。同步块执行时间较长。