0
点赞
收藏
分享

微信扫一扫

ReentrantLock 实现原理

ReentrantLock 实现原理_重入锁

使用 ​​synchronize​​ 来做同步处理时,锁的获取和释放都是隐式的,实现的原理是通过编译后加上不同的机器指令来实现。

而 ​ReentrantLock​​ 就是一个普通的类,它是基于 ​AQS(AbstractQueuedSynchronizer)​来实现的。

是一个重入锁:一个线程获得了锁之后仍然可以反复的加锁,不会出现自己阻塞自己的情况。

AQS​​ 是 ​Java​ 并发包里实现锁、同步的一个重要的基础框架。

锁类型

ReentrantLock 分为公平锁非公平锁,可以通过构造方法来指定具体类型:

  1.    //默认非公平锁
  2.    public ReentrantLock() {
  3.        sync = new NonfairSync();
  4.    }

  5.    //公平锁
  6.    public ReentrantLock(boolean fair) {
  7.        sync = fair ? new FairSync() : new NonfairSync();
  8.    }

默认一般使用非公平锁,它的效率和吞吐量都比公平锁高的多(后面会分析具体原因)。

获取锁

通常的使用方式如下:

  1.    private ReentrantLock lock = new ReentrantLock();
  2.    public void run() {
  3.        lock.lock();
  4.        try {
  5.            //do bussiness
  6.        } catch (InterruptedException e) {
  7.            e.printStackTrace();
  8.        } finally {
  9.            lock.unlock();
  10.        }
  11.    }

公平锁获取锁

首先看下获取锁的过程:

  1.    public void lock() {
  2.        sync.lock();
  3.    }

可以看到是使用 ​sync​​的方法,而这个方法是一个抽象方法,具体是由其子类( ​FairSync​)来实现的,以下是公平锁的实现:

  1.        final void lock() {
  2.            acquire(1);
  3.        }

  4.        //AbstractQueuedSynchronizer 中的 acquire()
  5.        public final void acquire(int arg) {
  6.        if (!tryAcquire(arg) &&
  7.            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
  8.            selfInterrupt();
  9.        }

第一步是尝试获取锁( ​tryAcquire(arg)​),这个也是由其子类实现:

  1.        protected final boolean tryAcquire(int acquires) {
  2.            final Thread current = Thread.currentThread();
  3.            int c = getState();
  4.            if (c == 0) {
  5.                if (!hasQueuedPredecessors() &&
  6.                    compareAndSetState(0, acquires)) {
  7.                    setExclusiveOwnerThread(current);
  8.                    return true;
  9.                }
  10.            }
  11.            else if (current == getExclusiveOwnerThread()) {
  12.                int nextc = c + acquires;
  13.                if (nextc < 0)
  14.                    throw new Error("Maximum lock count exceeded");
  15.                setState(nextc);
  16.                return true;
  17.            }
  18.            return false;
  19.        }
  20.    }

首先会判断 ​AQS​​ 中的 ​state​ 是否等于 0,0 表示目前没有其他线程获得锁,当前线程就可以尝试获取锁。

注意:尝试之前会利用 ​hasQueuedPredecessors()​ 方法来判断 AQS 的队列中中是否有其他线程,如果有则不会尝试获取锁(这是公平锁特有的情况)。

如果队列中没有线程就利用 CAS 来将 AQS 中的 state 修改为1,也就是获取锁,获取成功则将当前线程置为获得锁的独占线程( ​setExclusiveOwnerThread(current)​)。

如果 ​state​​ 大于 0 时,说明锁已经被获取了,则需要判断获取锁的线程是否为当前线程( ​ReentrantLock​​ 支持重入),是则需要将 ​state+1​,并将值更新。

写入队列

如果 ​tryAcquire(arg)​​ 获取锁失败,则需要用 ​addWaiter(Node.EXCLUSIVE)​ 将当前线程写入队列中。

写入之前需要将当前线程包装为一个 ​Node​​ 对象( ​addWaiter(Node.EXCLUSIVE)​)。

AQS 中的队列是由 Node 节点组成的双向链表实现的。

包装代码:

  1.    private Node addWaiter(Node mode) {
  2.        Node node = new Node(Thread.currentThread(), mode);
  3.        // Try the fast path of enq; backup to full enq on failure
  4.        Node pred = tail;
  5.        if (pred != null) {
  6.            node.prev = pred;
  7.            if (compareAndSetTail(pred, node)) {
  8.                pred.next = node;
  9.                return node;
  10.            }
  11.        }
  12.        enq(node);
  13.        return node;
  14.    }

首先判断队列是否为空,不为空时则将封装好的 ​Node​​ 利用 ​CAS​​ 写入队尾,如果出现并发写入失败就需要调用 ​enq(node);​ 来写入了。

  1.    private Node enq(final Node node) {
  2.        for (;;) {
  3.            Node t = tail;
  4.            if (t == null) { // Must initialize
  5.                if (compareAndSetHead(new Node()))
  6.                    tail = head;
  7.            } else {
  8.                node.prev = t;
  9.                if (compareAndSetTail(t, node)) {
  10.                    t.next = node;
  11.                    return t;
  12.                }
  13.            }
  14.        }
  15.    }

这个处理逻辑就相当于 ​自旋​​加上 ​CAS​ 保证一定能写入队列。

挂起等待线程

写入队列之后需要将当前线程挂起(利用 ​acquireQueued(addWaiter(Node.EXCLUSIVE),arg)​):

  1.    final boolean acquireQueued(final Node node, int arg) {
  2.        boolean failed = true;
  3.        try {
  4.            boolean interrupted = false;
  5.            for (;;) {
  6.                final Node p = node.predecessor();
  7.                if (p == head && tryAcquire(arg)) {
  8.                    setHead(node);
  9.                    p.next = null; // help GC
  10.                    failed = false;
  11.                    return interrupted;
  12.                }
  13.                if (shouldParkAfterFailedAcquire(p, node) &&
  14.                    parkAndCheckInterrupt())
  15.                    interrupted = true;
  16.            }
  17.        } finally {
  18.            if (failed)
  19.                cancelAcquire(node);
  20.        }
  21.    }

首先会根据 ​node.predecessor()​ 获取到上一个节点是否为头节点,如果是则尝试获取一次锁,获取成功就万事大吉了。

如果不是头节点,或者获取锁失败,则会根据上一个节点的 ​waitStatus​​ 状态来处理( ​shouldParkAfterFailedAcquire(p,node)​)。

waitStatus​ 用于记录当前节点的状态,如节点取消、节点等待等。

shouldParkAfterFailedAcquire(p,node)​​ 返回当前线程是否需要挂起,如果需要则调用 ​parkAndCheckInterrupt()​:

  1.    private final boolean parkAndCheckInterrupt() {
  2.        LockSupport.park(this);
  3.        return Thread.interrupted();
  4.    }

他是利用 ​LockSupport​​ 的 ​part​ 方法来挂起当前线程的,直到被唤醒。

非公平锁获取锁

公平锁与非公平锁的差异主要在获取锁:

公平锁就相当于买票,后来的人需要排到队尾依次买票,不能插队

而非公平锁则没有这些规则,是抢占模式,每来一个人不会去管队列如何,直接尝试获取锁。

非公平锁:

  1.        final void lock() {
  2.            //直接尝试获取锁
  3.            if (compareAndSetState(0, 1))
  4.                setExclusiveOwnerThread(Thread.currentThread());
  5.            else
  6.                acquire(1);
  7.        }

公平锁:

  1.        final void lock() {
  2.            acquire(1);
  3.        }

还要一个重要的区别是在尝试获取锁时 ​tryAcquire(arg)​,非公平锁是不需要判断队列中是否还有其他线程,也是直接尝试获取锁:

  1.        final boolean nonfairTryAcquire(int acquires) {
  2.            final Thread current = Thread.currentThread();
  3.            int c = getState();
  4.            if (c == 0) {
  5.                //没有 !hasQueuedPredecessors() 判断
  6.                if (compareAndSetState(0, acquires)) {
  7.                    setExclusiveOwnerThread(current);
  8.                    return true;
  9.                }
  10.            }
  11.            else if (current == getExclusiveOwnerThread()) {
  12.                int nextc = c + acquires;
  13.                if (nextc < 0) // overflow
  14.                    throw new Error("Maximum lock count exceeded");
  15.                setState(nextc);
  16.                return true;
  17.            }
  18.            return false;
  19.        }

释放锁

公平锁和非公平锁的释放流程都是一样的:

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

  4.    public final boolean release(int arg) {
  5.        if (tryRelease(arg)) {
  6.            Node h = head;
  7.            if (h != null && h.waitStatus != 0)
  8.                   //唤醒被挂起的线程
  9.                unparkSuccessor(h);
  10.            return true;
  11.        }
  12.        return false;
  13.    }

  14.    //尝试释放锁
  15.    protected final boolean tryRelease(int releases) {
  16.        int c = getState() - releases;
  17.        if (Thread.currentThread() != getExclusiveOwnerThread())
  18.            throw new IllegalMonitorStateException();
  19.        boolean free = false;
  20.        if (c == 0) {
  21.            free = true;
  22.            setExclusiveOwnerThread(null);
  23.        }
  24.        setState(c);
  25.        return free;
  26.    }        

首先会判断当前线程是否为获得锁的线程,由于是重入锁所以需要将 ​state​ 减到 0 才认为完全释放锁。

释放之后需要调用 ​unparkSuccessor(h)​ 来唤醒被挂起的线程。

总结

由于公平锁需要关心队列的情况,得按照队列里的先后顺序来获取锁(会造成大量的线程上下文切换),而非公平锁则没有这个限制。

所以也就能解释非公平锁的效率会被公平锁更高。

号外

最近在总结一些 Java 相关的知识点,感兴趣的朋友可以一起维护。

地址: https://github.com/crossoverJie/Java-Interview


举报

相关推荐

0 条评论