首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >bat面试必会的排他锁

bat面试必会的排他锁

作者头像
程序员小强
发布2019-08-16 10:48:21
4020
发布2019-08-16 10:48:21
举报

点击上方“小强的进阶之路”,选择“置顶或者星标”

你关注的就是我关心的!

预计阅读时间:11分钟
AQS的全称为(AbstractQueuedSynchronizer)

这个类在java.util.concurrent.locks下面, 这个类似乎很不容易看懂,因为它仅仅是提供了一系列公共的方法,让子类来调用。那么要理解意思,就得从子类下手,反过来看才容易看懂。

ReentrantLock排他锁

首先来看看ReentrantLock的构造方法,它的构造方法有两个

public ReentrantLock() {
    sync = new NonfairSync();
}

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

很显然,对象中有一个属性叫sync,有两种不同的实现类,默认是“NonfairSync”来实现,而另一个“FairSync”它们都是排它锁的内部类,不论用那一个都能实现排它锁,只是内部可能有点原理上的区别。先以“NonfairSync”类为例,它的lock()方法是如何实现的呢?

/**
 * Sync object for non-fair locks
 */
static final class NonfairSync extends Sync {
    private static final long serialVersionUID = 7316153563782823691L;

    /**
     * Performs lock.  Try immediate barge, backing up to normal
     * acquire on failure.
     */
    final void lock() {
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }

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

lock()方法先通过CAS尝试将状态从0修改为1。若直接修改成功,前提条件自然是锁的状态为0,则直接将线程的OWNER修改为当前线程,这是一种理想情况,如果并发粒度设置适当也是一种乐观情况。若上一个动作未成功,则会间接调用了acquire(1)来继续操作,这个acquire(int)方法就是在AbstractQueuedSynchronizer当中了。

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);
        return true;
    }
    return false;
}
  • 首先获取这个锁的状态,如果状态为0,则尝试设置状态为传入的参数(这里就是1),若设置成功就代表自己获取到了锁,返回true了。状态为0设置1的动作在外部就有做过一次,内部再一次做只是提升概率,而且这样的操作相对锁来讲不占开销。
  • 如果状态不是0,则判定当前线程是否为排它锁的Owner,如果是Owner则尝试将状态增加acquires(也就是增加1),如果这个状态值越界,则会抛出异常提示,若没有越界,将状态设置进去后返回true(实现了类似于偏向的功能,可重入,但是无需进一步征用)。
  • 如果状态不是0,且自身不是owner,则返回false。
public final void acquire(int arg) {
    if (!tryAcquire(arg) &&
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

tryAcquire()的调用判定中是通过if(!tryAcquire())作为第1个条件的,如果返回true,则判定就不会成立了,自然后面的acquireQueued动作就不会再执行了,如果发生这样的情况是最理想的。

无论多么乐观,征用是必然存在的,如果征用存在则owner自然不会是自己,tryAcquire()方法会返回false,接着就会再调用方法:acquireQueued(addWaiter(Node.EXCLUSIVE), arg)做相关的操作。

这里的Node.EXCLUSIVE是节点的类型, 从它的名称可以看到它的类型是排他的意思.接着调用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;
}

这里创建了一个Node的对象,将当前线程和传入的Node.EXCLUSIVE传入,也就是说Node节点理论上包含了这两项信息。

代码中的tail是AQS的一个属性,刚开始的时候肯定是为null,也就是不会进入第一层if判定的区域,而直接会进入enq(node)的代码,那么直接来看看enq(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;
            }
        }
    }
}

看到了tail就应该猜到了AQS是链表吧,没错,而且它还应该有一个head引用来指向链表的头节点,AQS在初始化的时候head、tail都是null,在运行时来回移动。

此时,我们最少至少知道AQS是一个基于状态(state)的链表管理方式。

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);
    }
}
  • 这里也是一个死循环,除非进入if(p == head && tryAcquire(arg))这个判定条件,而p为node.predcessor()得到,这个方法返回node节点的前一个节点,也就是说只有当前一个节点是head的时候,进一步尝试通过tryAcquire(arg)来征用才有机会成功。
  • tryAcquire(arg)这个方法我们前面介绍过,成立的条件为:锁的状态为0,且通过CAS尝试设置状态成功或线程的持有者本身是当前线程才会返回true
  • ○ 如果这个条件成功后,发生的几个动作包含:

(1) 首先调用setHead(Node)的操作,这个操作内部会将传入的node节点作为AQS的head所指向的节点。线程属性设置为空(因为现在已经获取到锁,不再需要记录下这个节点所对应的线程了),再将这个节点的perv引用赋值为null。

(2) 进一步将的前一个节点的next引用赋值为null。

private void setHead(Node node) {
    head = node;
    node.thread = null;
    node.prev = null;
}

相应的,可以自己看看FairSync实现类的lock方法,其实区别不大,有些细节上的区别可能会决定某些特定场景的需求。

End

本文参与 腾讯云自媒体分享计划,分享自微信公众号。
原始发表:2019-08-13,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 MoziInnovations 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体分享计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • 预计阅读时间:11分钟
  • AQS的全称为(AbstractQueuedSynchronizer)
  • ReentrantLock排他锁
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档