Java锁相关知识
Java锁相关知识
小胖子编程 发表于9个月前
Java锁相关知识
  • 发表于 9个月前
  • 阅读 2
  • 收藏 0
  • 点赞 0
  • 评论 0

移动开发云端新模式探索实践 >>>   

从ReentrantLock入手,学习Java锁相关知识

首先来看一下Java锁的使用

public static void main(String[] args) {
        ReentrantLock lock = new ReentrantLock();
        try {
            lock.lock();
            System.out.println("hello");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }

    }

ReentrantLock

先来看一下ReentrantLock类的构造器

private final Sync sync;

public ReentrantLock() {
        sync = new NonfairSync();
}
public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
}

默认构造器是创建一个NonfairSync类赋值给sync变量,另一个构造器传入一个boolean值来为sync变量赋值不同的对象,这里先不做深究。 先来看一下ReentrantLock加锁和解锁方法。

public void lock() {
        sync.lock();
}
public void unlock() {
        sync.release(1);
}

由此可见,ReentrantLock类实际的加锁解锁操作委托给了构造器创建的sync变量。 下面,让我们来看看sync是何方神圣。

abstract static class Sync extends AbstractQueuedSynchronizer {
      abstract void lock();
}

Sync是ReentrantLock的内部静态抽象类,继承了AbstractQueuedSynchronizer类,这里先不对AbstractQueuedSynchronizer进行展开。 Sync定义了抽象方法lock,那么ReentrantLock的lock方法,实际上由Sync的子类实现。

先分析默认构造器赋值的NonfairSync类实现的lock方法

 static final class NonfairSync extends Sync {
        final void lock() {
            if (compareAndSetState(0, 1)) //是否可以获得锁
                setExclusiveOwnerThread(Thread.currentThread()); //可以获得锁,将独享线程设为当前线程
            else
                acquire(1); //获得锁失败
        }
    }

追踪代码可知,compareAndSetState和acquire都是AbstractQueuedSynchronizer抽象类中实现的方法。 其加锁的实质,是依赖于AbstractQueuedSynchronizer类。 那么AbstractQueuedSynchronizer类究竟是什么呢?

AbstractQueuedSynchronizer,简称AQS,可以称之为同步器。 java.util.concurrent相关的锁机制,都是基于同步器实现的。

先看AQS的两个成员变量:

 private transient volatile Node head;
 private transient volatile Node tail;

显而易见,当线程并发获取锁失败时,AQS内部创建了一个FIFO队列(CLH队列)来存储获取等待锁竞争的线程。

下面来详细看一下Node结构

static final class Node {
 static final Node SHARED = new Node();
 static final Node EXCLUSIVE = null;

 static final int CANCELLED =  1;
 static final int SIGNAL    = -1;
 static final int CONDITION = -2;
 static final int PROPAGATE = -3;

 volatile int waitStatus;
 volatile Node prev;
 volatile Node next;
 volatile Thread thread;
 Node nextWaiter;

 final boolean isShared() {
         return nextWaiter == SHARED;
  }

 final Node predecessor() throws NullPointerException {
            Node p = prev;
            if (p == null)
                throw new NullPointerException();
            else
                return p;
  }

 Node() {    // Used to establish initial head or SHARED marker
 }

 Node(Thread thread, Node mode) {     // Used by addWaiter
            this.nextWaiter = mode;
            this.thread = thread;
 }

 Node(Thread thread, int waitStatus) { // Used by Condition
            this.waitStatus = waitStatus;
            this.thread = thread;
 }
}

先不详细解释Node的含义,回过头来看ReentrantLock中NonfairSync的lock方法:

final void lock() {
      if (compareAndSetState(0, 1))
           setExclusiveOwnerThread(Thread.currentThread());
      else
           acquire(1);
 }

先来看compareAndSetState方法,compareAndSetState是AQS实现的方法。

private static final Unsafe unsafe = Unsafe.getUnsafe();

stateOffset = unsafe.objectFieldOffset
              (AbstractQueuedSynchronizer.class.getDeclaredField("state"));

protected final boolean compareAndSetState(int expect, int update) {
      return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

其本质是调用了Unsafe的compareAndSwapInt方法

public final native boolean compareAndSwapInt(Object var1, long var2, int var4, int var5);

Unsafe的compareAndSwapInt方法是Native方法,若不追究其真正实现,理解到这里就差不多了。

---------------------------------------可以跳过------------------------------------------ compareAndSwapInt在Mac OS下的实现

UNSAFE_ENTRY(jboolean, Unsafe_CompareAndSwapInt(JNIEnv *env, jobject unsafe, jobject obj, jlong offset, jint e, jint x))
  UnsafeWrapper("Unsafe_CompareAndSwapInt");
  oop p = JNIHandles::resolve(obj); //获取该对象
  jint* addr = (jint *) index_oop_from_field_offset_long(p, offset); //根据该字段的偏移量得到内存地址
  return (jint)(Atomic::cmpxchg(x, addr, e)) == e; 
UNSAFE_END

其本质是调用Atomic::cmpxchg方法来实现原子操作

inline jint Atomic::cmpxchg(jint exchange_value, volatile jint* dest, jint compare_value) {
  unsigned int old_value;
  const uint64_t zero = 0;

  __asm__ __volatile__ (
    /* fence */
    strasm_sync
    /* simple guard */
    "   lwz     %[old_value], 0(%[dest])                \n"
    "   cmpw    %[compare_value], %[old_value]          \n"
    "   bne-    2f                                      \n"
    /* atomic loop */
    "1:                                                 \n"
    "   lwarx   %[old_value], %[dest], %[zero]          \n"
    "   cmpw    %[compare_value], %[old_value]          \n"
    "   bne-    2f                                      \n"
    "   stwcx.  %[exchange_value], %[dest], %[zero]     \n"
    "   bne-    1b                                      \n"
    /* acquire */
    strasm_sync
    /* exit */
    "2:                                                 \n"
    /* out */
    : [old_value]       "=&r"   (old_value),
                        "=m"    (*dest)
    /* in */
    : [dest]            "b"     (dest),
      [zero]            "r"     (zero),
      [compare_value]   "r"     (compare_value),
      [exchange_value]  "r"     (exchange_value),
                        "m"     (*dest)
    /* clobber */
    : "cc",
      "memory"
    );

  return (jint) old_value;
}

---------------------------------------完美的分割线------------------------------------

compareAndSwapInt其实质是CAS操作(关于CAS的相关知识,可以看看我写的另一篇文章CAS),来保证state的原子性。 当set值成功时,会返回true,若失败,则意味着已经有其它线程成功改变state。也就意味着获取锁失败。

继续看lock方法,当获取锁成功时,setExclusiveOwnerThread(Thread.currentThread())将当前线程设为排它锁当前占用的线程。 当compareAndSwapInt失败时,说明已经有线程抢先一步修改了state点的值,那么此时会执行acquire(1),acquire也是AQS实现的方法,那么acquire做了些什么呢?

public final void acquire(int arg) {
        if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
}

首先调用tryAcquire()方法来判断是否可获得锁,那么tryAcquire做了些什么?

 protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
}

可见,tryAcquire是一个真正实现的方法,这里使用了模板方法模式,由子类去实现相应的逻辑来判断是否可获得锁的判断,那么我们回过头来看看NonfairSync实现的tryAcquire方法。

  protected final boolean tryAcquire(int acquires) {
            return nonfairTryAcquire(acquires);
  }

其实质调用了Sync实现的nonfairTryAcquire方法

final boolean nonfairTryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState(); //获取当前state的值
            if (c == 0) { //若未被修改成1,则没有线程占用锁
                if (compareAndSetState(0, acquires)) { // 考虑到并发,用CAS更新state的值
                    setExclusiveOwnerThread(current);  //如果成功,则将当前线程设为排它锁当前占用的线程。
                    return true;
                }
            }
            else if (current == getExclusiveOwnerThread()) { //如果当前已经有线程占用了锁,那么判断是否是其本身占用了。
                int nextc = c + acquires; 
                if (nextc < 0) // overflow
                    throw new Error("Maximum lock count exceeded");
                setState(nextc); //如果是,则继续累加state的值,在这里也可以知道,ReentrantLock是可重入锁。传入1的含义也就明白了,每次占用加1次。
                return true;
            }
            return false; //如果已经有线程占用锁了,并且不是其本身,那么认为该线程获取锁失败。
        }

回过头看acquire方法, 当线程尝试获取锁成功时,则结束。 当线程尝试获取锁失败时,先调用addWaiter方法,将该线程存储到队列中,等待竞争锁

private Node addWaiter(Node mode) {
        Node node = new Node(Thread.currentThread(), mode); //先创建一个mode=Node.EXCLUSIVE的Node节点,将此节点加入到AQS维护的队列中。Node.EXCLUSIVE也可以看出来,accquire方法是实现排它锁的模板方法。
        Node pred = tail;
        if (pred != null) { //尝试一次快速加入队列,提高性能
            node.prev = pred;
            if (compareAndSetTail(pred, node)) { //与之前讲的compareAndSwapInt一样,以原子性修改tail节点
                pred.next = node;
                return node;
            }
        }
        enq(node); //若之前加入队列失败,则重新加入队列
        return node;
    }
private Node enq(final Node node) {
        for (;;) { //以自旋方式加入队列
            Node t = tail;
            if (t == null) { // 初始化队列
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
                node.prev = t; 
                if (compareAndSetTail(t, node)) {  //加入队列
                    t.next = node;
                    return t;
                }
            }
        }
}

加入AQS的队列之后,调用acquireQueued方法,让我们看看这个方法做了些什么

final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) { //自旋判断队列中能否竞争到锁
                final Node p = node.predecessor();  //获得该节点的前置节点
                if (p == head && tryAcquire(arg)) { //如果前置节点是头节点,则说明当前节点要被唤醒,认为当前节点有资格竞争锁,再一次尝试获取锁
                    setHead(node); //如果成功,则将头节点设为当前节点,标记该节点已获取锁,由此可知,头节点一定是获得锁的节点(或者是初始化的节点)
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&  //如果获取锁失败,则判断该节点是否需要被挂起
                    parkAndCheckInterrupt()) //挂起
                    interrupted = true;
            }
        } finally {
            if (failed) //如果程序抛出异常
                cancelAcquire(node); //那么取消该节点的占用(先不展开)
        }
}

先来看看shouldParkAfterFailedAcquire如何判断是否需要被挂起

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus; //获取前置节点的等待状态
        if (ws == Node.SIGNAL) //如果前置节点已经是唤醒状态,说明已经被之前修改过,那么该节点要等待前置节点来唤醒他
            return true; //那么该节点需要被挂起等待被唤醒
//其余状态则认为不需要被挂起,需要继续重试
        if (ws > 0) { //如果前置节点已经被取消了
            do {
                node.prev = pred = pred.prev; //那么一直往前,清除被取消的节点,直到找到未被取消的节点
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL); //将前置节点等待状态设为唤醒(需要前置节点来唤醒当前节点),自旋下一次该节点将被认定挂起
        }
        return false;
    }
private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this); //挂起当前线程
        return Thread.interrupted(); //返回线程中断标记,若执行到这一步,只有两种情况,一种是因为锁被释放而唤醒线程取消挂起,另一种是线程被中断而取消挂起。
    }

注意:因为在这里调用了Thread.interrupted(),若返回true,则证明该线程不是被唤醒,而是被中断才取消挂起的,此时interrupted被赋值为true,而Thread.interrupted()中断标记会被清除,所以为了避免中断标记被清除的问题,之前说的调用了selfInterrupt就是为了自己重新产生一个中断。由此可见,accquire也是对中断不敏感的。 ---------------------------------------可以跳过------------------------------------------ 那么是如何实现线程挂起的呢?这里使用了LockSupport辅助类,让我们来看看他干了些什么

public static void park(Object blocker) {
        Thread t = Thread.currentThread();
        setBlocker(t, blocker);
        UNSAFE.park(false, 0L);
        setBlocker(t, null);
}

setBlocker的意义是为了方便记录线程被阻塞时被谁阻塞的,用于线程监控和分析工具来定位原因的. 其实质还是调用了Unsafe的park方法

public native void park(boolean var1, long var2);

又是一个本地方法。有兴趣的可以研究一下本地实现。

UNSAFE_ENTRY(void, Unsafe_Park(JNIEnv *env, jobject unsafe, jboolean isAbsolute, jlong time))
  UnsafeWrapper("Unsafe_Park");
  EventThreadPark event;
#ifndef USDT2
  HS_DTRACE_PROBE3(hotspot, thread__park__begin, thread->parker(), (int) isAbsolute, time);
#else /* USDT2 */
   HOTSPOT_THREAD_PARK_BEGIN(
                             (uintptr_t) thread->parker(), (int) isAbsolute, time);
#endif /* USDT2 */
  JavaThreadParkedState jtps(thread, time != 0);
  thread->parker()->park(isAbsolute != 0, time);
#ifndef USDT2
  HS_DTRACE_PROBE1(hotspot, thread__park__end, thread->parker());
#else /* USDT2 */
  HOTSPOT_THREAD_PARK_END(
                          (uintptr_t) thread->parker());
#endif /* USDT2 */
  if (event.should_commit()) {
    oop obj = thread->current_park_blocker();
    event.set_klass((obj != NULL) ? obj->klass() : NULL);
    event.set_timeout(time);
    event.set_address((obj != NULL) ? (TYPE_ADDRESS) cast_from_oop<uintptr_t>(obj) : 0);
    event.commit();
  }
UNSAFE_END

---------------------------------------完美的分割线------------------------------------ 那么让我们来总结一下ReentrantLock的lock方法: 先通过CAS来判断是否可以获取锁,若获取成功,则标记当前线程占用锁资源。如果CAS获取失败,那么已经有线程抢先占用,那么调用AQS的accquire模板方法来获取一个许可。 那么accquire方法如何来或许一个许可呢,先调用tryAccquire方法来判断是否可以尝试获取许可,该逻辑交给子类自行去实现(一般就是对state进行CAS操作来判断是否可以获取锁资源,即获取许可),如果成功获取,那么线程获取锁资源成功,程序继续往下走,若尝试或许锁失败,由于ReentrantLock为了实现可重入的性质,那么先判断是否是获得锁的线程本身再次去尝试去获取锁,如果是,那么让state再加一,也就是标识着获取锁的次数。如果不是已经获取到锁的线程去尝试获取锁,那么,为了实现排它锁的性质,则认定获取锁失败。由AQS的addWaiter方法将尝试或许锁失败的线程加入到AQS维护的队列中。 调用acquireQueued再次去判断队列中线程是否可以获取锁资源,通过自旋的方式来判断线程是否可以获得锁资源,如果获取成功,则将队列头节点设为该节点(头节点为已经有资格获取锁资源或者为初始化节点),若获取失败,则判断是否需要挂起,真正的阻塞操作在挂起线程这边执行。至于如何判断是否需要挂起,则是判断前置非取消节点是否需要被唤醒(则认为前置节点有资格去竞争锁,该节点必然是需要等待前置节点操作结束)。取消挂起后,判断是是否是中断引起的,若是中断引起的,则重新产生一个中断。否则,则认为是被唤醒有资格获取锁,那么正常结束。

unlock

那么让我们再来看看unlock的实现

public void unlock() {
        sync.release(1);
}

其实质是调用了AQS的release方法,那让我们继续看看release方法

 public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
}

同样是一个模板方法模式,tryRelease由子类自己实现,尝试释放锁 那让我们来看看NonfairSync是如何尝试释放锁的,NonfairSync自身没有实现tryRelease方法,那么再去查看他的父类,可知,tryRelease由Sync实现

 protected final boolean tryRelease(int releases) {
            int c = getState() - releases;
            if (Thread.currentThread() != getExclusiveOwnerThread()) //如果当前释放锁的线程不是之前标记占用锁的线程,那么释放操作是异常的
                throw new IllegalMonitorStateException();
            boolean free = false;
            if (c == 0) { //因为可重入,state可能远大于1,所以一直释放直到c为0时,则认定之前重入的锁都释放完毕
                free = true; //此时标记可真正的释放锁资源
                setExclusiveOwnerThread(null); //清楚标记当前占用锁的线程
            }
            setState(c);
            return free; //返回是否可释放锁资源
}

tryRelease和tryAccquire是相对于的,主要是考虑到了可重入的性质。 继续看release方法,当判断可以释放锁资源以后, 获得头节点(即可以有资格获取锁资源的线程节点),判断不存在,则没有任何线程在竞争,那么直接返回结束。若存在头节点,并且头节点的等待状态被改变过(节点的等待状态何时会被改变?可以回过头看看acquire中判断线程是否需要被挂起,线程会将前置节点设为唤醒状态,当然也有可能是被取消的节点) 让我们来具体看看unparkSuccessor方法

 private void unparkSuccessor(Node node) {
        int ws = node.waitStatus; //获取头结点的等待状态
        if (ws < 0) 
            compareAndSetWaitStatus(node, ws, 0); //清除状态,即标记已经执行取消挂起操作
        Node s = node.next; //获得需要被唤醒的对象
        if (s == null || s.waitStatus > 0) {  //如果被唤醒的对象不存在,或者已经被取消
            s = null; 
            for (Node t = tail; t != null && t != node; t = t.prev) //从尾节点开始往前寻找未被取消的最后一个节点,也就是寻找头节点之后的未被取消的节点
//那么为什么要从尾节点开始找呢?
                if (t.waitStatus <= 0)
                    s = t;
        }
        if (s != null) //如果存在需要被唤醒的对象
            LockSupport.unpark(s.thread); //那么去唤醒他
    }

为何要从为节点开始往前找? 我们可以看看被取消的节点是如何产生的? 当节点竞争获取锁过程中抛异常时,会将此阶段设为被取消的节点,具体逻辑如下

 private void cancelAcquire(Node node) {
        if (node == null)
            return;
        node.thread = null; //将节点线程清空
        Node pred = node.prev;
        while (pred.waitStatus > 0)  //清除被取消的节点
            node.prev = pred = pred.prev;
        Node predNext = pred.next; //获取清除后的前置节点的后置节点
        node.waitStatus = Node.CANCELLED; //将该节点的等待状态设为被取消

        if (node == tail && compareAndSetTail(node, pred)) { //如果当前节点为尾节点,说明该线程节点是刚入队列的节点,把尾节点设为尾节点不被取消的前置节点
            compareAndSetNext(pred, predNext, null); //并把新的尾节点的后置节点置空  
        } else { //node不为尾节点或者把尾节点设为尾节点不被取消的前置节点失败时,说明有并发线程在同时获取锁资源导致
            int ws; 
            if (pred != head &&   //如果新的前置节点不是头结点
                ((ws = pred.waitStatus) == Node.SIGNAL || 
                 (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) && //将前置节点的等待状态设为唤醒状态
                pred.thread != null) { 
                Node next = node.next;   //获取当前节点的后置节点
                if (next != null && next.waitStatus <= 0)  //如果存在,则将前置节点的后置节点设为当前节点的后置节点,也就是废弃当前节点
                    compareAndSetNext(pred, predNext, next);
            } else {
                unparkSuccessor(node); //如果前置节点为头结点,那么原本需要被唤醒的节点被取消了,为了避免可能前置节点已经在触发unparkSuccessor的时候执行到取消挂起当前线程的时候,该线程节点还没有被设置成CANCELLED,那么就会去操作取消挂起这个线程的无效操作,所以需要再触发一次取消挂起操作,若头节点仍在占用锁,后置节点会重新挂起。
            }
            node.next = node; // help GC 将引用指向自身,方便GC回收
        }
    }
}

因为在设置被取消节点的时候,node.next = node操作使得next指针不可靠,如果从头节点开始找,可能会存在死循环的情况。 也正是因为next指针不可靠,所以依赖于prev指针,遇到判断waitStatus时,尽量清除被取消的节点,也由于next指针不可靠,所以清除时只需要考虑prev指针,也加强了性能。

至此,ReentrantLock的lock和unlock方法已经讲完了,另外的实现可以自行相应学习。 通过lock和unlock方法,我们学习到了AQS的accquire和release方法,这也是AQS为我们提供排它锁性质的同步器方法。 AQS还有实现了共享锁的方法,原理类似,有时间可以再细讲!

  • 打赏
  • 点赞
  • 收藏
  • 分享
共有 人打赏支持
粉丝 1
博文 9
码字总数 9126
×
小胖子编程
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
* 金额(元)
¥1 ¥5 ¥10 ¥20 其他金额
打赏人
留言
* 支付类型
微信扫码支付
打赏金额:
已支付成功
打赏金额: