Java语言中可以使用synchronized对一个对象或者方法进行加锁,然后互斥地执行synchronized包裹的代码块。synchronized代码块经过编译后会产生monitorenter和monitorexit字节码并分别作为代码块的开始和结束。上一篇提到,解释器执行monitorenter时会使用lock_object()锁住对象,lock_object()的具体实现如代码清单6-18所示:
代码清单6-18 lock_object()的实现
void InterpreterMacroAssembler::lock_object(Register lock_reg) {
// 如果强制使用重量级锁,lock_object()就不做优化了
if (UseHeavyMonitors) { ... } else {
...
// 将加锁对象放入obj_reg寄存器
movptr(obj_reg, Address(lock_reg, obj_offset));
// 如果开启偏向锁优化且偏向加锁成功,跳转到done,
// 否则跳到slow_case使用重量级锁
if (UseBiasedLocking) { biased_locking_enter(...); }
// 加载1到swap_reg
movl(swap_reg, (int32_t)1);
// 获取加锁对象的对象头,与1做位或运算,结果放入swap_reg
orptr(swap_reg,
Address(obj_reg, oopDesc::mark_offset_in_bytes()));
// 再将swap_reg保存到Displaced Headermovptr(Address(lock_reg, mark_offset), swap_reg);
// 使用对象头和swap_reg做比较,如果相等,将对象头替换为指向栈顶基本对象锁的指针,
// 加锁完成跳到done。否则将swap_reg设置为基本对象指针
lock();
cmpxchgptr(lock_reg,
Address(obj_reg, oopDesc::mark_offset_in_bytes()));
jcc(Assembler::zero, done);
// 加锁失败,再看看当前对象头是否已经是指向栈顶基本对象锁
const int zero_bits = LP64_ONLY(7) NOT_LP64(3);
subptr(swap_reg, rsp);
andptr(swap_reg, zero_bits - os::vm_page_size());
movptr(Address(lock_reg, mark_offset), swap_reg);
// 如果成功表示已经加过锁,跳到done完成。否则lock_object各种优化均失败,进入slow_
// case执行重量级锁
jcc(Assembler::zero, done);
// 重量级锁
bind(slow_case);
call_VM(...InterpreterRuntime::monitorenter);
bind(done);
}
}
如果用户强制使用重量级锁(-XX:+UseHeavyMonitors)那么使用lock_object()也无济于事。但默认情况下lock_object()会应用一系列优化措施:最开始尝试偏向锁,如果加锁失败则尝试基本对象锁,如果仍然失败再使用重量级锁。具体过程大致如下:不加锁→偏向锁→基本对象锁→重量级锁本节将详细讨论这三种锁优化技术,还会简单介绍x86引入的硬件事务内存锁。
锁优化的第一个尝试是偏向锁。如果开启-XX:+UseBiasedLocking偏向锁优化标志,虚拟机将尝试用偏向锁操作免除加锁同步带来的性能惩罚。偏向锁会记录第一次获取该锁对象的线程的指针,然后将它记录在对象头中,并修改对应的位。此时偏向锁偏向于该线程。接下来如果同一个线程在同一个对象上执行同步操作,那么这些操作无须任何原子指令,完全消除了后续加锁、解锁的开销。但是只要有其他线程尝试获取这个锁,偏向模式就会立即结束,虚拟机会撤销偏向,后续加锁、解锁则使用基本对象锁。
由于历史原因,在多个线程上使用对应的某个对象并进行大量同步操作时,与普通锁相比,偏向锁的性能有明显提升,但是在今天,这些性能提升变得不那么明显。现代处理器的原子操作比以前开销小,另外,由于偏向锁优化针对的应用程序一般都是那些老的、过时的应用程序,它们均使用Java早期的Collection API如Vector、Hashtable,这些类的每个操作都需要同步,而现在的应用程序,在单线程中一般使用非同步的HashMap、ArrayList,在多线程中使用更高效的并发数据结构,所以偏向锁对于现在的应用程序起到的优化效果甚微。除此之外,偏向锁的实现也相当复杂,阻碍了HotSpot VM开发者对代码各个部分的理解,也阻碍了HotSpot VM同步模块的设计变更。因此JEP 374提议在JDK15之后默认关闭偏向锁,并逐渐移除它。
如果偏向锁获取失败,虚拟机将尝试基本对象锁。前面提到在lock_object()调用前,栈上monitor区存在一个基本对象锁,包含锁住的对象和BasicLock,BasicLock又包含Displaced Header。虚拟机会尝试获取锁住的对象的对象头然后与1做位或操作(lock_object->mark() | 1),并将获得的结果放入rax寄存器和栈顶Displaced Header。接下来使用原子CAS指令比较rax寄存器和对象头,如果相等,说明对象没有加锁,可以将对象头替换为指向栈顶基本对象锁的指针和00轻量级锁模式。如果不相等,此时CAS操作会将对象头放入rax寄存器,然后查看对象头是否已经指向栈顶指针,即是否已经加过锁。若两次判断都失败,lock_object()膨胀为重量级锁ObjectMonitor。上述完整的加锁流程如图6-4所示。
图6-4是lock_object()的代码逻辑。对象头与1位或操作其实就是判断对象尾部2位以确认是否加锁。第3章曾提到32位和64位的对象头,它们的尾部有2位的锁模式。当锁模式为01时表示未被锁定,此时lock_obj->mark() == (lock_obj->mark()|1),对象头被替换为指向栈上基本对象锁的指针。基本对象锁总是机器位对齐,它的最后两位是00,而锁模式为00时表示已上锁。
如果上述操作都失败,虚拟机将会使用重量级锁。与Object.wait/notify等方法相同,重量级锁会调用runtime/synchronizer的ObjectSynchronizer,它封装了一些逻辑,如对象锁的分配和释放、对象头的改变等,然后由这些函数代理ObjectMonitor执行wait/notify等底层操作。
ObjectMonitor即重量级锁底层实现,与Monitor类似,ObjectMonitor也有cxq和EntryList的概念,不过ObjectMonitor的实现相对来说更为复杂,如代码清单6-19所示:
代码清单6-19 ObjectMonitor加锁解锁逻辑
void ObjectMonitor::enter(TRAPS) {
// CAS抢锁,如果当前线程抢到锁则直接返回
Thread * const Self = THREAD;
void * cur = Atomic::cmpxchg(Self, &_owner, (void*)NULL);
if (cur == NULL) {
return;
}
// 否则CAS返回_owner给cur,_owner的值可能是线程指针,也可能是基本对象锁
// 检查_owner是否为当前线程指针,如果是则当前线程再次加锁(递归计数加一)
if (cur == Self) {_recursions++;
return;
}
// 检查_owner是否为位于当前线程栈上的基本对象锁,如果是则递归计数加一以加锁
if (Self->is_lock_owned ((address)cur)) {
_recursions = 1;
_owner = Self;
return;
}
// 否则当前对象锁的_owner是其他线程或者位于其他线程栈上的基本对象锁
// 尝试自旋来和其他线程竞争该锁
Self->_Stalled = intptr_t(this);
if (TrySpin(Self) > 0) {
Self->_Stalled = 0;
return;
}
// 如果自旋竞争失败
JavaThread * jt = (JavaThread *) Self;
Atomic::inc(&_count);
{
// 改变当前线程状态,使其阻塞在对象锁上
...
EnterI(THREAD);
...
// 阻塞结束,线程继续执行exit(false, Self);
...
}
Atomic::dec(&_count);
Self->_Stalled = 0;
}
void ObjectMonitor::exit(bool not_suspended, TRAPS) {
Thread * const Self = THREAD;
// 如果_owner不是当前线程
if (THREAD != _owner) {
...
}
// 否则_owner是当前线程,或者当前线程栈上的基本对象锁
// 如果已经加过锁,递归计数减一即可
if (_recursions != 0) {
_recursions--;
return;
}
_Responsible = NULL;
// 由于当前线程没有递归加锁,同时又是对象锁的持有者,这意味着当前线程执行对象锁的exit,
// 同时还需要找到下一个待唤醒的线程,因为如果当前线程结束了同步执行又没有唤醒其他线程,
// 那么其他线程会无限等待下去
for (;;) {
// 将对象锁持有者置空
OrderAccess::release_store(&_owner, (void*)NULL);
OrderAccess::storeload();// 如果没有其他线程竞争对象锁,直接返回
if ((intptr_t(_EntryList)|intptr_t(_cxq)) == 0 || _succ != NULL){
return;
}
if (!Atomic::replace_if_null(THREAD, &_owner)) {
return;
}
// 如果EntryList中存在等待对象锁的线程
ObjectWaiter * w = NULL;
w = _EntryList;
if (w != NULL) {
ExitEpilog(Self, w);
return;
}
// cxq中存在等待对象锁的线程,将线程从cxq转移到EntryList
// ---- 1. 保存cxq
w = _cxq;
if (w == NULL) continue;
// ---- 2. 将cxq置空
for (;;) {
ObjectWaiter * u = Atomic::cmpxchg(NULL, &_cxq, w);
if (u == w) break;
w = u;
}
// ---- 3.将cxq转移到EntryList
_EntryList = w;// 将EntryList中的所有线程设置为TS_ENTER
ObjectWaiter * q = NULL;
ObjectWaiter * p;
for (p = w; p != NULL; p = p->_next) {
p->TState = ObjectWaiter::TS_ENTER;
p->_prev = q;
q = p;
}
if (_succ != NULL) continue;
// 唤醒EntryList的第一个线程
w = _EntryList;
if (w != NULL) {
ExitEpilog(Self, w);
return;
}
}
}
获取对象锁的核心逻辑是首先尝试使用CAS获取锁(设置_owner),如果失败再和其他线程正常竞争对象锁,并在竞争失败的情况下阻塞。
释放对象锁只需要检查当前线程是否持锁,如果持锁(且没有多次获取过,即递归计数为0)则释放锁(设置_owner为NULL),同时如果对象锁已经存在其他等待获取的线程,挑选一个等待对象锁的线程唤醒即可。
从因特尔微架构Haswell开始,增加了事务同步扩展指令集,该指令集包括硬件锁消除和受限事务内存(Restricted TransactionalMemory,RTM)。下面详细介绍RTM如何从硬件上支持程序执行事务代码。
RTM使用硬件指令实现。xbegin和xend限定了事务代码块的范围,两者结合相当于monitorenter和monitorexit。如果在事务代码块执行过程中没有异常发生,寄存器和内存的修改都会在xend执行时提交。xabort可以用于显式地终止事务的执行,xtest检查EIP/RIP是否位于事务代码块。前文提到过锁的膨胀过程大致如下:
不加锁→偏向锁→基本对象锁→重量级锁
如果开启-XX:+UseRTMLocking,经过C2编译后的代码的加锁过程会多一个RTM加锁代码:
无锁→基本对象锁→重量级锁的RTM加锁→重量级锁
如果同时开启-XX:+UseRTMLocking和-XX:+UseRTMForStackLocks,加锁过程会增加两步:
无锁→基本对象锁的RTM加锁→基本对象锁→重量级锁的RTM加锁→重量级锁
RTM的关键是无数据竞争。当没有数据竞争时,只要多个线程访问xbegin和xend限定事务代码中的同一个内存位置且没有写操作,那么硬件允许多个线程同时并行执行完事务,即使monitor代码段的语义是互斥执行。但是当发生数据竞争时,事务执行会失败,且事务终止的开销和事务重试的开销不容忽视。可见,RTM从实现到工业应用还有很长的一段路要走。
本文就是愿天堂没有BUG给大家分享的内容,大家有收获的话可以分享下,想学习更多的话可以到微信公众号里找我,我等你哦。