美文网首页
java多线程-6-ReentrantLock

java多线程-6-ReentrantLock

作者: 宠辱不惊的咸鱼 | 来源:发表于2019-09-30 09:06 被阅读0次

概述

  • 基于AQS(AbstractQueuedSynchronizer,concurrent包中的一个抽象类)
  • AQS基于CAS(Compare and Swap,一种并发技术),CAS在java中靠Unsafe实现(native,依靠C++和汇编)
  • Unsafe
    • Access restriction: The type 'Unsafe' is not API (restriction on required library 'C:\Program Files\Java\jdk1.8.0_111\jre\lib\rt.jar')
    • 已测试,确实跑不动
    • @CallerSensitive,被注解的方法,需要方法调用者具备一定权限
      • bootstrap class loader加载的类可调用
      • extension class loader加载的类可调用
private final Sync sync; // 持有Sync,锁的功能来自于Sync
private volatile int state; // Sync中属性
// 默认非公平锁
public ReentrantLock() {
    sync = new NonfairSync();
}
// 实际调用sync的方法,Sync是个抽象类,实现类有FairSync和NonfairSync,默认非公平
lock.lock()
    sync.lock()
// NonfairSync实现
NonfairSync
    final void lock() {
        if (compareAndSetState(0, 1)) // 试着抢一把(这就是非公平的特点)
            setExclusiveOwnerThread(Thread.currentThread()); // 设置当前线程独占锁
        else
            acquire(1); // 抢夺不成,排队去(公平锁直接来这)
    }

源码解析 - lock()

thread-003.jpg

第1个线程

  • compareAndSetState检查当前lock的state属性
  • 采用unsafe.compareAndSwapInt(this, stateOffset, expect, update)
    • 检查this对象的偏移地址为stateOffset的属性
      • stateOffset:unsafe.objectFieldOffset(AbstractQueuedSynchronizer.class.getDeclaredField("state"))
    • 是否与expect一致
      • 若是一致,就更新为update
      • 意味着,原state值为0才表示这个lock没有被别人持有,当前线程才可以改值
  • NonfairSync持有一个从AbstractOwnableSynchronizer继承过来的exclusiveOwnerThread属性,表示独占锁线程
  • 第一个获取锁的线程做了两件事情
    • 设置AQS(即Sync)的state为1
    • 设置exclusiveOwnerThread为当前线程

第2个线程

  • 线程2也尝试获取同一个锁,在线程1没有释放锁的情况下必然是行不通的,所以线程2就要阻塞

  • 那么,线程2如何被阻塞


    thread-004.jpg
  • 线程2执行compareAndSetState(0, 1)返回false

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

  • 先走第一个判断条件尝试获取锁
  • 如果获取失败,走第二个判断条件添加FIFO同步队列
  • 先看下tryAcquire方法做了什么?
protected final boolean tryAcquire(int acquires) {
    return nonfairTryAcquire(acquires);
}

final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        if (compareAndSetState(0, acquires)) {
            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); //这里无需CAS,偏向锁
        return true;
    }
    return false;
}
  • state是volatile的,对线程2具有可见性
  • 线程2拿到state,再次判断一下能否持有锁(可能线程1同步代码执行得比较快,这会儿已经释放了锁),不可以就返回false
  • 一个锁最多可重入Integer.MAX_VALUE次,也就是2147483647
  • 上述第二个判断条件中addWaiter逻辑
private Node addWaiter(Node mode) {
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    enq(node);
    return node;
}

private Node enq(final Node node) {
    for (;;) {
        Node t = tail;
        if (t == null) { // Must initialize
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
            node.prev = t;
            if (compareAndSetTail(t, node)) {
                t.next = node;
                return t;
            }
        }
    }
}
  • 先用当前线程创建一个Node
  • 判断tail有没有,有的话就以compareAndSetTail方式将当前node插在tail后面,并将node谁为tail
    • 若是tail不存在,进入enq(node)
      • 以compareAndSetHead的方式创建head
      • 将当前node作为head.next,以compareAndSetTail方式将node设为tail
    • 若是因为别的线程也在插入导致插入失败,则进入enq(node)一直插入直到成功
  • head是个dummy节点,不对应线程
  • 上述第二个判断条件中的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);
    }
}
  • 如果线程是双向队列第一个真Node(前面还有一个dummy head),那么尝试获取锁
    • 如果依旧失败,调用AQS的shouldParkAfterFailedAcquire(p, node)
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus; // node的前置节点
    if (ws == Node.SIGNAL)
        /*
         * This node has already set status asking a release
         * to signal it, so it can safely park.
         */
        return true;
    if (ws > 0) {
        /*
         * Predecessor was cancelled. Skip over predecessors and
         * indicate retry.
         */
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        /*
         * waitStatus must be 0 or PROPAGATE.  Indicate that we
         * need a signal, but don't park yet.  Caller will need to
         * retry to make sure it cannot acquire before parking.
         */
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}
  • head的waitStatus是0
  • 把head的waitStatus设置为Noed.SIGNAL即-1,并返回false
  • 下一次for循环,还是先尝试获取锁,不成功
  • 继续走shouldParkAfterFailedAcquire,此时waitStatus为-1,返回true
  • 执行parkAndCheckInterrupt
private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted(); // 返回中断标识,并清除
}

public static void park(Object blocker) {
    Thread t = Thread.currentThread();
    setBlocker(t, blocker);
    UNSAFE.park(false, 0L); // 进入阻塞,醒来时也在这醒来
    setBlocker(t, null);
}
  • 至此,lock()告一段落

源码解析 - unlock()

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

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
protected final boolean tryRelease(int releases) {
    int c = getState() - releases;
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {
        free = true;
        setExclusiveOwnerThread(null);
    }
    setState(c);
    return free;
}
  • 只有c等于0才会让free=true,即重入情况下需要多次释放
  • 若c为0,置exclusiveOwnerThread为null,表示没有线程占有锁,返回true
  • head不为空,且waitStatus为-1,所以执行unparkSuccessor(h)
private void unparkSuccessor(Node node) {
    /*
     * If status is negative (i.e., possibly needing signal) try
     * to clear in anticipation of signalling.  It is OK if this
     * fails or if status is changed by waiting thread.
     */
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);

    /*
     * Thread to unpark is held in successor, which is normally
     * just the next node.  But if cancelled or apparently null,
     * traverse backwards from tail to find the actual
     * non-cancelled successor.
     */
    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);
}
  • s就是线程2的Node,不为null,并且s.waitStatus是0(未被赋过值)
  • 线程2被unpark
  • 有一个很重要的问题是:锁被解了怎么保证FIFO队列减少一个Node呢?回到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 // 猜测是便于node本身之后的回收
                failed = false;
                return interrupted;
            }
            if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}
  • 线程2被唤醒后,依然进行for循环
  • 线程2所在Node的前驱节点p其实就是那个dummy head
  • 线程2尝试tryAcquire,成功
  • 线程2节点置为head
  • head这个局部变量在之前enq结束时就被回收,指向原head节点那块内存的引用就只剩下线程2节点
  • 在setHead(node)中,node.prev被置空,意味着指向原head内存引用全部解除,这样它就被GC自动回收了
  • 然后setHead中很关键的一句,node.thread=null,其实就是把现任head(即线程2节点)变成一个dummy head
private void setHead(Node node) {
    head = node;
    node.thread = null;
    node.prev = null;
}
  • 为什么需要dummy head
    • 因为一个线程随时可能因为中断而取消,而取消的话,Node自然就要被GC了,也就意味着head可能就没了。那GC前必然要把头Node的后继Node变为一个新的头而且要应对多种情况,这样就很麻烦。(虽然这个解释有点勉强,但貌似也可以接受)用一个没有thread的Node作为头,相当于起了一个引导作用,因为head没有线程,自然也不会被取消。
  • 看一下上面unparkSuccessor的14行~20行,是为了应对在head的下一个node可能已被取消的情况;从尾到头遍历,找出离head最近的一个真实等待的node,对这个node进行unpark操作
  • lockInterruptibly作用
    • 获取锁阶段可以被外界interrupt,这样就可以在某种死锁的情况下获得退出
    • 记得,它的作用不是在线程运行过程中接受interrupt并处理

相关文章

  • java多线程-6-ReentrantLock

    概述 基于AQS(AbstractQueuedSynchronizer,concurrent包中的一个抽象类) A...

  • 带你搞懂Java多线程(五)

    带你搞懂Java多线程(一)带你搞懂Java多线程(二)带你搞懂Java多线程(三)带你搞懂Java多线程(四) ...

  • 带你搞懂Java多线程(六)

    带你搞懂Java多线程(一)带你搞懂Java多线程(二)带你搞懂Java多线程(三)带你搞懂Java多线程(四)带...

  • Java多线程目录

    Java多线程目录 Java多线程1 线程基础Java多线程2 多个线程之间共享数据Java多线程3 原子性操作类...

  • java多线程--Callable

    **移步[java多线程系列文章]Java多线程(二十二)---LockSupport工具Java 停止线程 一、...

  • android 多线程 — 线程的面试题和答案

    这里都是我从各个地方找来的资料,鸣谢: Java多线程干货系列—(一)Java多线程基础 JAVA多线程和并发基础...

  • 5月份第一周学习安排

    学习内容: java多线程及线程同步的方法(使用) java多线程各种同步方法的原理和优缺点 java多线程设计模...

  • 带你搞懂Java多线程(四)

    带你搞懂Java多线程(一)带你搞懂Java多线程(二)带你搞懂Java多线程(三) 什么是线程间的协作 线程之间...

  • Java基础(六)

    多线程 Java多线程并发 1.1 JAVA 并发知识库 1.2 JAVA 线程实现/创建方式 1.2.1 继承 ...

  • (五) volatile关键字

    Java多线程目录 1 背景 理解Java多线程的内存抽象逻辑请阅读java多线程内存模型,当代操作系统,处理器为...

网友评论

      本文标题:java多线程-6-ReentrantLock

      本文链接:https://www.haomeiwen.com/subject/lwrcpctx.html