深入源码解析synchronized关键字

# 写在文章开头

因为之前关于synchronized关键字底层工作机制理解有所偏差，简单查阅了一下其底层的实现，遂以此文简单记录一下jdk8版本synchronized关键字的设计理念和实现，希望对你有所帮助。

我是 sharkchili ，CSDN Java 领域博客专家，mini-redis的作者，我想写一些有意思的东西，希望对你有帮助，如果你想实时收到我写的硬核的文章也欢迎你关注我的公众号： 写代码的SharkChili 。

同时也非常欢迎你star我的开源项目mini-redis:https://github.com/shark-ctrl/mini-redis

# 基于objectMonitor深入剖析synchronized关键字

# 宏观了解synchronized同步机制

我们先从全局的维度了解一下synchronized关键字的实现，假设我们现在有三个线程尝试争抢对象锁，此时线程0最先通过CAS机制将锁的持有者标识为自己并成功获取锁，对应的线程1和线程2则CAS失败。此阶段线程0就可以操作锁内部对应的临界资源，即synchronized底层锁持有者标识就是_owner字段指向线程0：

线程1和线程2在竞争失败后，会先尝试自旋获取锁，如果自旋失败则会进入阻塞队列中等待，直到锁释放，而这个队列可以是_cxq或者_EntryList。这里笔者为了简单直观地展现这一过程，将竞争失败的线程直接放入_EntryList队列来演示这一过程：

我们再聊聊锁重入的情况，例如我们线程0之前上锁调用synchronized修饰的是function1，内部又调用function2，对应关键字底层也就是objectMonitor对象，也就是通过一个_recursions字段对重入次数进行累加，这也就意味着线程0进行锁释放时必须释放两次：

public class Example1 {

    public synchronized void function1(){
        Console.log("function1");
        function2();
    }
  
    public synchronized void function2(){
        Console.log("function2");
      
    }
  
   
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14

然后就是挂起锁释放的场景，线程0期间因为特定原因，主动挂起调用wait方法，此时就需要完成锁释放操作，对应底层操作则是将其存入等待队列_WaitSet中，并将_EntryList中等待的线程唤醒尝试竞争锁(注意，不一定完全是_EntryList，这里更多是为了举例做一个普适的说明)：

sync-source-code-3.drawio

后续线程0被notify或者notifyall唤醒之后，还会再次回到_EntryList竞争锁。

自此我们从宏观的角度了解了synchronized底层的工作机制，对应的我们也通过源码的角度给出了上述概念中的几个成员字段，整体和上述表述类似

_count：等待获取该锁的线程数，通常为_cxq和_EntryList队列中的线程数总和
_waiters：调用wait方法进入等待状态的线程数
_owner：指向当前持有锁的线程
_recursions：线程重入次数，用于记录当前线程对同一把锁的重入次数

对应我们也给出ObjectMonitor结构体定义：

ObjectMonitor() {
    _header       = NULL;
    _count        = 0;//抢占该锁的线程数即 _cxq.size + EntryList.size
    _waiters      = 0,//处于等待状态的线程数
    _recursions   = 0;//线程重入次数
    _object       = NULL;
    _owner        = NULL;//当前锁即ObjectMonitor的拥有者
    _WaitSet      = NULL;//处于wait状态的线程会存入到该队列中
    _WaitSetLock  = 0 ;
    _Responsible  = NULL ;
    _succ         = NULL ;
    _cxq          = NULL ;//多线程竞争时会进入该队列中
    FreeNext      = NULL ;
    _EntryList    = NULL ;//处于block等待锁状态的线程会进入该队列
    _SpinFreq     = 0 ;
    _SpinClock    = 0 ;
    OwnerIsThread = 0 ;
    _previous_owner_tid = 0;
  }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

上述说明我们也可以在源码objectMonitor.cpp中的注释中查阅到，感兴趣的读者可以翻阅这段注释了解具体的实现细节，其大体意思与上述描述基本一致。对应的，笔者简单概括如下：

线程必须通过cas将owner从null变为自己才算成功获得锁
任意时刻线程因阻塞等待必须处于cxq或EntryList队列中，或者因wait调用进入WaitSet队列
持有锁线程释放后，其必须唤醒EntryList一个候选线程争抢锁
......

# 详解并发锁资源竞争

线程基于synchronized关键字争抢锁资源的本质实现就是objectMonitor.cpp的enter方法，其上锁步骤严格：

尝试cas将null设置为当前线程的指针，如果cas方法返回null则说明自己获取锁成功直接返回，反之进入步骤2
如果非空但返回指针地址是自己，说明当前线程之前已经获取过一个这把锁，本次为锁重入，直接累加重入次数_recursions后返回，反之进入步骤3
判断当前持有锁的线程是否是这把锁的轻量级锁持有者，如果是，则执行一次锁升级将其升级为重量级锁，并将owner对象中栈上BasicLockObject地址转为完整的thread指针，反之进入步骤4
上述步骤都不成功，则说明当前锁被其他线程持有，尝试自旋几次获取这把锁如果成功则直接返回，反之进入步骤5
为了避免非必要的CPU自旋开销，累加count表示当前获取锁失败的线程数加一，再次自旋尝试几次
几次尝试还是无果，将当前线程封装为等待节点存入等待队列，直到被唤醒拿到锁后退出
通过步骤6的方式拿到锁的线程，会通过cas的方式完成等待线程数count值扣减并退出

对应的我们给出了这段代码的流程图，读者可参考上述说明理解：

有了上述的基础之后，我们就可以查阅objectMonitor.cpp上锁逻辑enter的实现，逻辑和上述说明基本一致，我们也不难看出对应jdk8版本的synchronized关键字内置的优化，即核心理念就是尽可能利用cas竞争锁资源，明确感知到竞争激烈之后主动将其挂起，再利用等待通知模型唤醒线程竞争资源：

void ATTR ObjectMonitor::enter(TRAPS) {
 //获取到当前线程的指针
  Thread * const Self = THREAD ;
  void * cur ;
  //尝试CAS的方式将owner从null设置为当前线程
  cur = Atomic::cmpxchg_ptr (Self, &_owner, NULL) ;
  //如果返回null，说明当前线程是第一个获取到这把锁的，直接返回
  if (cur == NULL) {
    //......
     return ;
  }
  //非null且指针就是自己说明是重入，直接累加重入次数_recursions后返回
  if (cur == Self) {
     //......
     _recursions ++ ;
     return ;
  }
  //如果当前线程是轻量级锁的持有则，则进入该逻辑进行锁升级
  if (Self->is_lock_owned ((address)cur)) {
    //......
    //将owner从线程栈上的BasicLockObject地址转换为完整的Thread指针
    _owner = Self ;
    OwnerIsThread = 1 ;
    return ;
  }

  // We've encountered genuine contention.
  //当前锁被其他线程占有，需要抢占
  assert (Self->_Stalled == 0, "invariant") ;
  //记录需要抢占的线程monitor指针
  Self->_Stalled = intptr_t(this) ;


  //尝试自旋获取锁，成功后返回
  if (Knob_SpinEarly && TrySpin (Self) > 0) {
     //......
     return ;
  }
//......
  //经过上述步骤还是没有抢到锁，原子累加count值，即增加一个等待获取锁的线程
  Atomic::inc_ptr(&_count);

  EventJavaMonitorEnter event;

		//......
    for (;;) {
      jt->set_suspend_equivalent();
      // cleared by handle_special_suspend_equivalent_condition()
      // or java_suspend_self()
      //尝试CAS或者自旋等方式获取锁，若失败则通过头插法将线程存入cxq队列中，等待后续通知唤醒
      EnterI (THREAD) ;

      if (!ExitSuspendEquivalent(jt)) break ;

      //......
      //完成操作后，锁释放通知其他线程上锁
      exit (false, Self) ;

      jt->java_suspend_self();
    }
    Self->set_current_pending_monitor(NULL);

   
    // acquire it.
  }
  //原子扣减count告知等待锁的线程少一个
  Atomic::dec_ptr(&_count);
 //......
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

# 锁释放流程解析

对于锁释放流程，整体逻辑比较简单，对应的执行步骤为：

查看当前线程是否是owner指针指向的线程，如果不是，则存在锁升级但是当前线程是这把锁轻量级时期持有者的情况，则进一步判断线程是否是轻量级锁持有者，如果是则进入步骤2
查看_recursions是否不为0，若不为0则说明是重入，扣减_recursions后返回
上述检查无误，将锁直接释放，然后进入步骤4
当前线程进行cas尝试将owner从null设置为自己以查看是否存在其他线程竞争，如果失败则说明锁被其他线程持有，不需要执行后续唤醒线程的任务直接返回，反之进入步骤5
来到步骤5，这也就意味着当前监视锁没有被其它线程获取，当前线程需要执行唤醒等待线程的任务，唤醒线程不一定是从EntryList也可能从cxq队列，对应的唤醒策略由QMode决定，具体规则如下：

1. QMode为2直接唤醒cxq队列首元素线程，让其竞争监视锁
2. QMode为3，则查看cxq队列是否非空，若明确非空则将其追加到EntryList中，若EntryList为空则直接将cxq队列设置为EntryList，然后将首元素唤醒
3. QMode为4，则将EntryList追加到cxq队列后，然后让cxq队列作为EntryList首元素，将首个线程唤醒竞争监视锁


/************************* 执行到步骤4和5，说明代码中的逻辑判断明确EntryList为空，对应不同的QMode规则为**************/


4. QMode为1，将cxq队列倒叙排列后作为EntryList，并将首元素唤醒
5. QMode为0或2，将cxq直接封装为ObjectWaiter作为EntryList，并将首元素唤醒

1
2
3
4
5
6
7
8
9
10

特殊步骤：如果发现w为空则说明cxq和EntryList队列都为空，则重新从循环头开始执行正确的退出协议

对应如下是笔者整理的流程图，读者可以基于上述说明进行梳理归纳：

对应位于objectMonitor.cpp的exit函数就是我们释放锁的逻辑，这里笔者整理出核心的代码段，读者可结合注释更进一步理解：

void ATTR ObjectMonitor::exit(bool not_suspended, TRAPS) {
   Thread * Self = THREAD ;
   //如果当前线程不等于_owner则进入当前逻辑
   if (THREAD != _owner) {
     if (THREAD->is_lock_owned((address) _owner)) {//如果当前线程持有这把锁的轻量级锁，则将其设置为重量级锁并让_owner指向完整的线程地址，然后执行后续的锁释放逻辑
      //......
       assert (_recursions == 0, "invariant") ;
       _owner = THREAD ;
       _recursions = 0 ;
       OwnerIsThread = 1 ;
     } else {
     //......
       /*
     
       JNI（Java Native Interface）本地代码中可能出现的监视器（即 synchronized 锁）使用不平衡的情况，
       比如进入 monitor 但没有正确退出，或者退出没有对应的进入。当检测到这种异常情况时，应该抛出 Java 的
       IllegalMonitorStateException 异常来通知开发者。
       */
       TEVENT (Exit - Throw IMSX) ;
       assert(false, "Non-balanced monitor enter/exit!");
       if (false) {
          THROW(vmSymbols::java_lang_IllegalMonitorStateException());
       }
       return;
     }
   }

   //_recursions不为0，说明此前线程重入过，完成_recursions扣减后返回，不重置owner指针
   if (_recursions != 0) {
     _recursions--;        // this is simple recursive enter
     TEVENT (Inflated exit - recursive) ;
     return ;
   }

   //......

   for (;;) {
      assert (THREAD == _owner, "invariant") ;


      if (Knob_ExitPolicy == 0) {
        //......

         //将ownder设置为null，即释放锁
         OrderAccess::release_store_ptr (&_owner, NULL) ;   // drop the lock
         OrderAccess::storeload() ;                         // See if we need to wake a successor
        //......
         //cas尝试一下线程从null设置为自己，如果非空则说明锁被其他线程持有，不执行后续唤醒其他线程的逻辑
         if (Atomic::cmpxchg_ptr (THREAD, &_owner, NULL) != NULL) {
            return ;
         }
         TEVENT (Exit - Reacquired) ;
      } else {
         if ((intptr_t(_EntryList)|intptr_t(_cxq)) == 0 || _succ != NULL) {
          //释放锁
            OrderAccess::release_store_ptr (&_owner, NULL) ;   // drop the lock
           //......
            //尝试cas上锁，如果不成功则说明当前锁被其他线程持有，则当前线程不负责后续唤醒等待线程的职责，直接返回
            if (Atomic::cmpxchg_ptr (THREAD, &_owner, NULL) != NULL) {
               TEVENT (Inflated exit - reacquired succeeded) ;
               return ;
            }
            TEVENT (Inflated exit - reacquired failed) ;
         } else {
            TEVENT (Inflated exit - complex egress) ;
         }
      }

    //......

      ObjectWaiter * w = NULL ;
      int QMode = Knob_QMode ;
	
      if (QMode == 2 && _cxq != NULL) {//QMode为2且cxq队列非空，直接唤醒cxq队列首元素直接返回
        
          w = _cxq ;
          assert (w != NULL, "invariant") ;
          assert (w->TState == ObjectWaiter::TS_CXQ, "Invariant") ;
          //传入_cxq队列首元素地址，唤醒对应线程去竞争锁资源
          ExitEpilog (Self, w) ;
          return ;
      }

      if (QMode == 3 && _cxq != NULL) {
       
        //
          // 通过CAS将_cxq设置为空
          // 将w（即原有的cxq队列）转换为双向链表的waiter
          // 将最近到达的线程（RATs）追加到EntryList
          // TODO：将EntryList组织为循环双向链表（CDLL），这样我们就能在常量时间内定位到尾部。
          // 把cxq挂到_EntryList队列上，后续从_EntryList中唤醒
          // 查找EntryList的尾部
          // 如果EntryList为空，则将w设为EntryList
          // 否则将w链接到EntryList尾部
          w = _cxq ;
          for (;;) {
             assert (w != NULL, "Invariant") ;
             //通过cas把cxq设置为空
             ObjectWaiter * u = (ObjectWaiter *) Atomic::cmpxchg_ptr (NULL, &_cxq, w) ;
             if (u == w) break ;
             w = u ;
          }
      

          ObjectWaiter * q = NULL ;
          ObjectWaiter * p ;
          //将w即原有cxq队列转为一个双向链表的waiter
          for (p = w ; p != NULL ; p = p->_next) {
              guarantee (p->TState == ObjectWaiter::TS_CXQ, "Invariant") ;
              p->TState = ObjectWaiter::TS_ENTER ;
              p->_prev = q ;
              q = p ;
          }

     
          //如果_EntryList非空的话直接将其追加到cxq队列，反之cxq队列直接作为_EntryList
          ObjectWaiter * Tail ;
          for (Tail = _EntryList ; Tail != NULL && Tail->_next != NULL ; Tail = Tail->_next) ;
          if (Tail == NULL) {
              _EntryList = w ;
          } else {
              Tail->_next = w ;
              w->_prev = Tail ;
          }
        

          // Fall thru into code that tries to wake a successor from EntryList
      }

      if (QMode == 4 && _cxq != NULL) {//如果QMode且_cxq非空，则把_EntryList挂到cxq上
          // Aggressively drain cxq into EntryList at the first opportunity.
     
          //w指针指向cxq队列后，通过cas将cxq指针置空
          w = _cxq ;
          for (;;) {
             assert (w != NULL, "Invariant") ;
             ObjectWaiter * u = (ObjectWaiter *) Atomic::cmpxchg_ptr (NULL, &_cxq, w) ;
             if (u == w) break ;
             w = u ;
          }
          assert (w != NULL              , "invariant") ;
          //将w(原cxq队列封装为等待队列)
          ObjectWaiter * q = NULL ;
          ObjectWaiter * p ;
          for (p = w ; p != NULL ; p = p->_next) {
              guarantee (p->TState == ObjectWaiter::TS_CXQ, "Invariant") ;
              p->TState = ObjectWaiter::TS_ENTER ;
              p->_prev = q ;
              q = p ;
          }

          // Prepend the RATs to the EntryList
          //将_EntryList追加到w(原cxq队列)
          if (_EntryList != NULL) {
              q->_next = _EntryList ;
              _EntryList->_prev = q ;
          }
          _EntryList = w ;

          // Fall thru into code that tries to wake a successor from EntryList
      }

      w = _EntryList  ;
      if (w != NULL) {//从_EntryList找到节点唤醒指定线程返回
      
          assert (w->TState == ObjectWaiter::TS_ENTER, "invariant") ;
          ExitEpilog (Self, w) ;
          return ;
      }

   
      w = _cxq ;
      //如果发现w为空则说明cxq和entrylist队列都为空，则重新从循环头开始执行正确的退出协议
      if (w == NULL) continue ;

   
      for (;;) {
          assert (w != NULL, "Invariant") ;
          ObjectWaiter * u = (ObjectWaiter *) Atomic::cmpxchg_ptr (NULL, &_cxq, w) ;
          if (u == w) break ;
          w = u ;
      }
  

      if (QMode == 1) {
         // QMode == 1 : drain cxq to EntryList, reversing order
         // We also reverse the order of the list.
         //当 QMode 设置为 1 时，
         //ObjectMonitor 会将 cxq（竞争队列）中的所有等待线程移动到 EntryList（入口列表）中，
         // 但在这个过程中会将线程的顺序颠倒。这种设计可能是为了实现某种调度策略，比如让最后到达的线程优先获得锁（LIFO - 后进先出）
         ObjectWaiter * s = NULL ;
         ObjectWaiter * t = w ;
         ObjectWaiter * u = NULL ;
         while (t != NULL) { //将cxq队列元素倒叙排列一下
             guarantee (t->TState == ObjectWaiter::TS_CXQ, "invariant") ;
             t->TState = ObjectWaiter::TS_ENTER ;
             u = t->_next ;
             t->_prev = u ;
             t->_next = s ;
             s = t;
             t = u ;
         }
         _EntryList  = s ;
         assert (s != NULL, "invariant") ;
      } else {
         // QMode == 0 or QMode == 2 直接顺序生成等待节点
         _EntryList = w ;
         ObjectWaiter * q = NULL ;
         ObjectWaiter * p ;
         for (p = w ; p != NULL ; p = p->_next) {
             guarantee (p->TState == ObjectWaiter::TS_CXQ, "Invariant") ;
             p->TState = ObjectWaiter::TS_ENTER ;
             p->_prev = q ;
             q = p ;
         }
      }

   
      //拿到首元素唤醒，让其尝试竞争监视锁资源
      w = _EntryList  ;
      if (w != NULL) {
          guarantee (w->TState == ObjectWaiter::TS_ENTER, "invariant") ;
          ExitEpilog (Self, w) ;
          return ;
      }
   }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226

# 挂起等待的wait调用

线程调用wait方法时就会释放锁进入等待队列，等待超时this.wait(5000L);或者被唤醒后重新竞争监视锁资源，对应的处理步骤为：

进行必要的中断检查，如果发现调用wait时线程正在处理中断，则直接抛出中断异常，反之进入步骤2
将节点封装为等待节点
自旋获取等待队列的锁，将线程存入等待队列
保存好当前线程的_recursions重入次数后(便于后续线程再次持有锁之后重入和释放锁次数一致性)，将锁的_recursions重置，调用exit退出监视锁
按照wait传参进行有限或者无限时间的等待，直到被唤醒
唤醒后，查看自己当前状态如果处于TS_RUN则直接尝试竞争锁，如果是TS_ENTER或者TS_CXQ则进行必要的自旋尝试竞争锁后挂起等待
上锁成功后，恢复锁重入次数，扣减_waiters告知等待线程减少一个
操作临界资源

这段逻辑比较简单，读者可直接基于上述说明查看wait代码的底层实现：

void ObjectMonitor::wait(jlong millis, bool interruptible, TRAPS) {
  //Self指向当前线程的指针
   Thread * const Self = THREAD ;
   //......

   // check for a pending interrupt
   // 检查是否存在待处理的中断，若明确处理中断则直接抛出中断异常
   if (interruptible && Thread::is_interrupted(Self, true) && !HAS_PENDING_EXCEPTION) {
     //......
     TEVENT (Wait - Throw IEX) ;
     THROW(vmSymbols::java_lang_InterruptedException());
     return ;
   }

	 //......
   //将当前节点封装为wait
   ObjectWaiter node(Self);
   node.TState = ObjectWaiter::TS_WAIT ;
   Self->_ParkEvent->reset() ;
   //......
   //自旋获取waitSet锁将节点存入waiterSet然后释放锁
   Thread::SpinAcquire (&_WaitSetLock, "WaitSet - add") ;
   AddWaiter (&node) ;
   Thread::SpinRelease (&_WaitSetLock) ;

  
   intptr_t save = _recursions; // record the old recursion count 保存旧的递归计数，便于后续唤醒恢复
   _waiters++;                  // increment the number of waiters 增加等待者数量
   _recursions = 0;             // set the recursion level to be 1 将递归级别设置为1
   
   exit (true, Self) ;                    // exit the monitor 退出监视器
 

   //......

   int ret = OS_OK ;
   int WasNotified = 0 ;
   //获取当前线程的park挂起
   { // State transition wrappers
     OSThread* osthread = Self->osthread();
     OSThreadWaitState osts(osthread, true);
     {
      //......

       if (interruptible && (Thread::is_interrupted(THREAD, false) || HAS_PENDING_EXCEPTION)) {
           // Intentionally empty
           // 故意为空
       } else
       //按照指定的传入等待时间参数等待，若没设置时间则无限期挂起等待直到唤醒
       if (node._notified == 0) {
         if (millis <= 0) {
            Self->_ParkEvent->park () ;
         } else {
            ret = Self->_ParkEvent->park (millis) ;
         }
       }

      //......

     } // Exit thread safepoint: transition _thread_blocked -> _thread_in_vm
     //......


   

     //自旋获取锁将线程中等待节点移除，并将其状态设置为TS_RUN，尝试争抢锁
     if (node.TState == ObjectWaiter::TS_WAIT) {
         Thread::SpinAcquire (&_WaitSetLock, "WaitSet - unlink") ;
         if (node.TState == ObjectWaiter::TS_WAIT) {
            DequeueSpecificWaiter (&node) ;       // unlink from WaitSet
            // 从WaitSet中解除链接
            assert(node._notified == 0, "invariant");
            node.TState = ObjectWaiter::TS_RUN ;
         }
         Thread::SpinRelease (&_WaitSetLock) ;
     }

   

     //......
     ObjectWaiter::TStates v = node.TState ;
     //查看被唤醒的线程当前处于什么状态，如果状态为TS_RUN则调用enter直接去竞争锁，反之说明节点在等待队列中，进行必要的自旋竞争后进入cxq或者entrylist中等待唤醒
     if (v == ObjectWaiter::TS_RUN) {
         enter (Self) ;
     } else {
         guarantee (v == ObjectWaiter::TS_ENTER || v == ObjectWaiter::TS_CXQ, "invariant") ;
         ReenterI (Self, &node) ;
         node.wait_reenter_end(this);
     }

     //......
   } 
	 //......
   _recursions = save;     // restore the old recursion count 恢复旧的递归计数

   _waiters--;             // decrement the number of waiters 减少等待者数量



  //......

  
}

# 通知唤醒的notify操作

在分析wait源码的时候，笔者提及wait唤醒的队列可能在cxq或者在entryList队列中，本质上是notify操作的等待队列处理策略，当线程通过notify操作尝试通知处于waitset中的线程时，其底层的notify调用整体执行步骤为：

判断_WaitSet是否非空，若非空执行步骤2
通过自旋获取_WaitSet的锁，着手处理_WaitSet中的线程
上锁成功后，按照配置的策略处理_WaitSet中的元素，具体策略为：

策略0：如果EntryList非空的话，则将等待者添加到EntryList的头部
策略1：将等待者添加到EntryList的尾部
策略2：将等待者添加到cxq的头部
策略3：将等待者添加到cxq的尾部
其他:  直接唤醒WaitSet中的首个线程，让其竞争监视锁

1
2
3
4
5

释放WaitSet锁

notify源码逻辑比较清晰简单，读者可结合笔者的注释自行阅读了解一下细节，并串联上述的说明完成逻辑闭环：

void ObjectMonitor::notify(TRAPS) {
  CHECK_OWNER();

  //_WaitSet为空直接返回
  if (_WaitSet == NULL) {
     TEVENT (Empty-Notify) ;
     // 记录空通知事件
     return ;
  }
  DTRACE_MONITOR_PROBE(notify, this, object(), THREAD);

  int Policy = Knob_MoveNotifyee ;

  //通过自旋获取等待队列的锁
  Thread::SpinAcquire (&_WaitSetLock, "WaitSet - notify") ;
  //拿到waitSet
  ObjectWaiter * iterator = DequeueWaiter() ;
  // 从等待队列中取出一个等待者
  if (iterator != NULL) {
     TEVENT (Notify1 - Transfer) ;
     guarantee (iterator->TState == ObjectWaiter::TS_WAIT, "invariant") ;
     guarantee (iterator->_notified == 0, "invariant") ;
     if (Policy != 4) {
        iterator->TState = ObjectWaiter::TS_ENTER ;
        // 将等待者状态设置为进入状态
     }
     iterator->_notified = 1 ;

     Thread * Self = THREAD;
     iterator->_notifier_tid = Self->osthread()->thread_id();
   
     //定位到等待队列
     ObjectWaiter * List = _EntryList ;
     // 获取EntryList
     if (List != NULL) {
        assert (List->_prev == NULL, "invariant") ;
        assert (List->TState == ObjectWaiter::TS_ENTER, "invariant") ;
        assert (List != iterator, "invariant") ;
     }
     // 策略0：如果EntryList非空的话，则将等待者添加到EntryList的头部
     if (Policy == 0) {       // prepend to EntryList
         if (List == NULL) {
             iterator->_next = iterator->_prev = NULL ;
             _EntryList = iterator ;
           
         } else {
             List->_prev = iterator ;
             iterator->_next = List ;
             iterator->_prev = NULL ;
             _EntryList = iterator ;
           
        }
     } else
     if (Policy == 1) {      // append to EntryList
         // 策略1：将等待者添加到EntryList的尾部
         if (List == NULL) {
             iterator->_next = iterator->_prev = NULL ;
             _EntryList = iterator ;
             // 如果EntryList为空，则直接将等待者设为EntryList
         } else {
       
            ObjectWaiter * Tail ;
            for (Tail = List ; Tail->_next != NULL ; Tail = Tail->_next) ;
            // 查找EntryList的尾部
            assert (Tail != NULL && Tail->_next == NULL, "invariant") ;
            // 断言：尾部不为NULL且下一个元素为NULL
            Tail->_next = iterator ;
            iterator->_prev = Tail ;
            iterator->_next = NULL ;
            // 将等待者添加到EntryList的尾部
        }
     } else
     //策略2：将等待者添加到cxq的头部
     if (Policy == 2) {      // prepend to cxq
       
         // prepend to cxq
         if (List == NULL) {
             iterator->_next = iterator->_prev = NULL ;
             _EntryList = iterator ;
             // 如果EntryList为空，则直接将等待者设为EntryList
         } else {
            iterator->TState = ObjectWaiter::TS_CXQ ;
            // 将等待者状态设置为CXQ状态
            for (;;) {
                ObjectWaiter * Front = _cxq ;
                iterator->_next = Front ;
                if (Atomic::cmpxchg_ptr (iterator, &_cxq, Front) == Front) {
                    break ;
                    // 原子性地将等待者添加到cxq的头部
                }
            }
         }
     } else
     //策略3：将等待者添加到cxq的尾部
     if (Policy == 3) {      // append to cxq
      
        iterator->TState = ObjectWaiter::TS_CXQ ;
        // 将等待者状态设置为CXQ状态
        for (;;) {
            ObjectWaiter * Tail ;
            Tail = _cxq ;
            if (Tail == NULL) {
                iterator->_next = NULL ;
                if (Atomic::cmpxchg_ptr (iterator, &_cxq, NULL) == NULL) {
                   break ;
                   // 如果cxq为空，则原子性地将等待者设为cxq
                }
            } else {
                while (Tail->_next != NULL) Tail = Tail->_next ;
                // 查找cxq的尾部
                Tail->_next = iterator ;
                iterator->_prev = Tail ;
                iterator->_next = NULL ;
                break ;
                // 将等待者添加到cxq的尾部
            }
        }
     } else {
        ParkEvent * ev = iterator->_event ;
        // 获取等待者的事件
        iterator->TState = ObjectWaiter::TS_RUN ;
        // 将等待者状态设置为运行状态
        OrderAccess::fence() ;
        // 内存屏障
        ev->unpark() ;
        // 直接唤醒等待者
     }

   
  }
  // 释放WaitSet锁
  Thread::SpinRelease (&_WaitSetLock) ;
  

  //......
}

# 关于synchronized更进一步的理解

# 为什么锁释放之后还要进行一次cas上锁

上文提到执行exit方法时，线程通过release_store_ptr完成锁释放之后，会尝试cas再次尝试将锁的owner指向自己，这样做的目的是什么呢？

//释放锁
OrderAccess::release_store_ptr (&_owner, NULL) ;   // drop the lock
//......
//cas上锁
if (Atomic::cmpxchg_ptr (THREAD, &_owner, NULL) != NULL) {
            return ;
}
//执行后续唤醒逻辑

1
2
3
4
5
6
7
8

这本质上是针对cpu时间片的利用和调度效率上的一种优化，即当前线程完成锁释放后，尽可能利用当前线程还在使用时间片的时刻，尝试cas上锁，如果上锁成功则说明当前锁没有人获取，则执行后续唤醒等待线程的逻辑，由此尽可能利用cpu时间片避免搁浅问题(锁释放了所有线程还是处于park状态)：

# 自旋重试的性能压榨

竞争锁资源的enter方法在明确几次cas和自旋失败后，会将其添加到cxq队列中，但设计者并不会直接添加，而是尽可能利用每一次机会，利用我们将线程封装为等待节点会通过cas存入等待队列，为了尽可能利用每一次cpu时间片，再进行cas自旋入队时，设计者会利用每次cas失败的时机再次尝试自旋上锁，这本质就是对于入队激烈竞争的一种临时规避，尽可能利用这个间隙再次获取锁资源：

对应笔者也贴出这段核心代码段：

for (;;) {
      //当前节点指向cxq队列的首元素
        node._next = nxt = _cxq ;
        //通过cas将当前元素设置为cxq队列的首元素，若成功则退出循环
        if (Atomic::cmpxchg_ptr (&node, &_cxq, nxt) == nxt) break ;

        // Interference - the CAS failed because _cxq changed.  Just retry.
        // As an optional optimization we retry the lock.
        //若cas失败则再次尝试自旋上锁，尽可能利用cpu执行开销
        if (TryLock (Self) > 0) {
            assert (_succ != Self         , "invariant") ;
            assert (_owner == Self        , "invariant") ;
            assert (_Responsible != Self  , "invariant") ;
            return ;
        }
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

# 不同唤醒策略的设计理念

锁释放时，源码中对于线程的唤醒策略给出了大体5种策略，实际上这也是对系统资源规律的把握和不同场景效率的优化，对应的笔者结合之前整理的几种策略进行说明：

1. QMode为2直接唤醒cxq队列首元素线程，让其竞争监视锁，这种方式避免cxq转移到EntryList的开销，直接让cxq队列元素唤醒，适用于追求高并发和极致的场景，不需要考虑公平性
2. QMode为3，则查看cxq队列是否非空，若明确非空则将其追加到EntryList尾部，若EntryList为空则直接将cxq队列设置为EntryList，然后将首元素唤醒，因为线程竞争不到锁资源之后是采用头插法存入cxq队列，所以为了避免cxq反序追加到EntryList这个开销，该策略是直接将cxq队列追加到EntryList上，这本质上就是一种公平性和性能的折中
3. QMode为4，则将EntryList追加到cxq队列后，然后让cxq队列作为EntryList首元素，将首个线程唤醒竞争监视锁，这种方案是考虑到最新的线程会位于cxq队列队首，所以cpu缓存中可能存有这个线程的缓存数据，因此优先唤醒该线程保证高效完成计算
/************************* 执行到步骤4和5，说明代码中的逻辑判断明确EntryList为空，对应不同的QMode规则为**************/
4. QMode为1，将cxq队列倒序排列后作为EntryList，并将首元素唤醒，这种就是将头插法的cxq队列顺序排列追加到EntryList，严格保证调度的公平性，避免线程饥饿
5. QMode为0（默认策略），将cxq直接封装为ObjectWaiter作为EntryList，绕过两个列表关联直接将首元素唤醒，适用于高并发和高性能的场景

1
2
3
4
5
6

# 为什么wait获取锁采用自旋而非重量级锁

_WaitSetLock保护的等待队列本质上基本只有监视器所有者进行访问，个别情况由于超时中断返回操作队列的情况，所以竞争不算激烈，所以在操作等待队列时上锁的方式是采用自旋而非重量级锁：

   Thread::SpinAcquire (&_WaitSetLock, "WaitSet - add") ;
   AddWaiter (&node) ;
   Thread::SpinRelease (&_WaitSetLock) ;

1
2
3

# 小结

本文针对synchronized关键字上锁、释放锁、等待通知等操作的底层机制和优化理念进行了深入分析和讲解，希望对你有所启发。

同时也非常欢迎你star我的开源项目mini-redis:https://github.com/shark-ctrl/mini-redis

因为近期收到很多读者的私信，所以也专门创建了一个交流群，感兴趣的读者可以通过上方的公众号获取笔者的联系方式完成好友添加，点击备注 “加群” 即可和笔者和笔者的朋友们进行深入交流。

# 参考

内置锁(ObjectMonitor):https://www.cnblogs.com/hongdada/p/14513036.html (opens new window)

Java面试常见问题：Monitor对象是什么？:https://zhuanlan.zhihu.com/p/356010805 (opens new window)

Java并发基石——所谓"阻塞"：Object Monitor和AQS（1）:https://blog.csdn.net/yinwenjie/article/details/84922958 (opens new window)

Java多线程：objectMonitor源码解读（3）:https://juejin.cn/post/7255230505409527863 (opens new window)

ObjectMonitor:https://www.jianshu.com/p/c0854b241c2d (opens new window)

Java Synchronized 重量级锁原理深入剖析上(互斥篇) :https://blog.csdn.net/wekajava/article/details/120306478 (opens new window)

编辑

← 浅谈Java线程池中拒绝策略与流控的艺术《JavaScript高级程序设计》笔记→