转载

AQS:JAVA经典之锁实现算法(二)-Condition

使用过 ReentrantLock 的盆友应该也知道 Condition 的存在。先讲解下它存在的意义:就是仿照实现 Object 类的 wait signal signallAll 等函数功能的。

这里引申一个面试常问到的问题: wait 会释放锁, sleep 不会。

  • Condition 的通常使用场景是这样的: 生产者消费者模型,假设生产者只有在生产队列为空时才进行生产,则代码类似如下:
Condition emptyCondition = ReentrantLock.newCondition();
Runnable consumer = new Runnable() {
  public void run() {
    if(queue.isEmpty()) {
      emptyCondition.signal();  // emptyObj.notify();
    } else {
      consumer.consume();
    }
  }
}
Runnable provider = new Runnable() {
  public void run() {
    emptyCondition.wait();  // emptyObj.wait();
    providerInstance.produce();
  }
}
复制代码

所以我们可以知道 Condition 设计的意义了。下面我们来讲解下其实现原理。

一:实现概况

还记得在 AQS:JAVA经典之锁实现算法(一) 提到的锁实现的 Sync Queue 吗? Condition 的实现是类似的原理: 每个 AQS 里有x(视你 newCondition 几次)个 Condition Queue ,它的结点类也是 AQS 内部类 NodeNode 里有一个 nextWaiter ,指向下一个在同一 Condition Queue 里的 Node 。 结构如下图:

AQS:JAVA经典之锁实现算法(二)-Condition
  • 首先明确下是, condition.wait 一定是在成功 lock 的线程里调用才有效,不然不符合逻辑,同时也会抛出 IlleagleMornitorException
  • 获取锁的线程处于 Sync Queue 的队首,当调用 condition.wait 时,该线程会释放锁(即将 AQSstate 置为0),同时唤醒后继结点,后继结点在 acquire 的循环里会成功获取锁,然后将自己所在结点置为队首,然后开始自己线程自己的业务代码。 这个过程看下图:
    AQS:JAVA经典之锁实现算法(二)-Condition
AQS:JAVA经典之锁实现算法(二)-Condition
  • 当waiter_1收到相应 conditionsignal 后,在 Condition Queue 中的 Node 会从 Condition Queue 中出队,进入 Sync Queue 队列,开始它的锁竞争的过程。 过程看下图:
AQS:JAVA经典之锁实现算法(二)-Condition
AQS:JAVA经典之锁实现算法(二)-Condition

所以,这里可以看出来,即使是被 signal 了,被 signal 的线程也不是直接就开始跑,而是再次进入 Sync Queue 开始竞争锁而已。这里的这个逻辑,跟 Object.wait Object.signal 也是完全一样的。

二:代码实现原理

我们先看一段运用到 condition 的代码案例: 假设生成者在生产队列 queue 为空时 emptyCondition.signal 才进行生产操作

ReentrantLock locker = new ReentrantLock();
Condition emptyCondition = locker.newCondition();

Runnable consumer = new Runnable() {
  public void run() {
    locker.lock();
    if (queue.isEmpty()) {
      emptyCondition.signal();
    } else {
      ...
    }
    locker.unlock();
  }
};

Runnable producer = new Runnable() {
  public void run() {
    locker.lock();
    emptyCondition.wait();
    // 开始生产
    ...
    locker.unlock();
  }
}
复制代码

我们从消费者一步一步走,拟定如下这样一套线程切换逻辑:

producer#lock
consumer#lock
producer#await
consumer#signal
consumer#unlock
producer#unlock

(先从 Sync Queue Condition Queue 图解讲一遍,然后对应图解,对着代码撸一遍)

  • producer#lock

生产者直接获取锁成功,入队 Sync Queue ,位队首

AQS:JAVA经典之锁实现算法(二)-Condition

consumer#lock

消费者竞争锁失败,进入 Sync Queue 等待获取锁

AQS:JAVA经典之锁实现算法(二)-Condition
  • producer#await

生产者进入等待,释放锁,出 Sync Queue ,进入 Condition Queue ,等待 emptyCondition 来唤醒。

AQS:JAVA经典之锁实现算法(二)-Condition
  • consumer#signal

消费者唤起生产者,生产者 consumernodeCondition Queue 转移到 Sync Queue 开始竞争锁。

AQS:JAVA经典之锁实现算法(二)-Condition
  • consumer.unlock

consumer 释放锁后, consumernodeSync Queue 出队,释放 state ,唤醒后继结点 provider#nodeprovider 抢占到锁。

AQS:JAVA经典之锁实现算法(二)-Condition
  • provider#unlock

这里就没有啥好说的了。

当然,我为了讲解过程,像在锁被第一次成功获取的时候,逻辑上虽然并不是直接进入 Sync Queue 我也给讲解成直接进入 Sync Queue 了,这是为了缩减边边角角的小逻辑,讲清楚主线逻辑。大家看明白主逻辑,然后再自己去撸一遍,就融会贯通了。

三:代码撸一把

  • provider.lock

final void lock() {
            // 这就直接获取锁成功了,没有else的逻辑了
            if (compareAndSetState(0, 1))
                // 这个方法是AQS类用来设置拥有锁的线程实例
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        }
复制代码
  • consumer#lock

final void lock() {
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            // consumer.lock就要走这里了,因为上面的compareAndSetState
            // 返回false
            else
                acquire(1);
        }
复制代码
protected final boolean compareAndSetState(int expect, int update) {
        // 楼下这个是CAS原理进行值修改,CAS就对比乐观锁来,
        // 这里想要修改this这个对象的state字段,如果state是expect
        // 则修改至update,返回true;否则false。我们知道provider.lock
        // 已经将state 改为非0值了,所以这里肯定失败啦
        return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
    }
复制代码
  • provider#await

先简单看下 Condition 类对象结构

public class ConditionObject implements Condition, java.io.Serializable {
        private static final long serialVersionUID = 1173984872572414699L;
        /** First node of condition queue. */
        private transient Node firstWaiter;
        /** Last node of condition queue. */
        private transient Node lastWaiter;
}
...
复制代码

一个 Condition 对象就是一条链队,头尾结点在 Condition 的内部字段指定 firstWaiter lastWaiter

await 方法

public final void await() throws InterruptedException {
            // 因为await是响应中断的等待,这里就是检验下,
            // 通常而言,凡是throws InterruptedException的,
            // 开头基本都是这句
            if (Thread.interrupted())
                throw new InterruptedException();
            // 这里是向condition queue中插入一个node,并返回之,
            // 插入了这个node,就代表当前线程在condition queue
            // 中开始等待了
            Node node = addConditionWaiter();
            // 这个是AQS释放锁方法,加个fully,就是用来将多次
            // 获取锁一次性都释放掉,然后将锁获取次数返回,
            // 留着后面signal后成功获取锁的时候,还要加锁同样的
            // 次数。
            // !!!同时注意,这里唤醒了后继结点!后集结点就继续开始
            // 竞争锁,就是在acquire那个自旋方法里,记得吗
            // 不记得去看看文章(一)
            int savedState = fullyRelease(node);
            // 记录当前线程中断的标记
            int interruptMode = 0;
            // 判断当前的node是否已经转移到sync queue里了。
            // 转移了,说明这个node已经开始竞争锁了,不用再等待
            // 唤醒了,没转,继续自旋
            while (!isOnSyncQueue(node)) {
                // 这里把当前线程给挂起了
                LockSupport.park(this);
                // 这里的方法checkxxx就是用来检查waiting自旋期间,线程有没有
                // interrupt掉。因为await方法是响应线程中断的。
                // 若interrupt了,则在checkxxx方法里,会将node转移到
                // sync Queue中,去竞争,不要担心,因为同时
                // 会设置interruptMode,在最后会根据其值抛Interrupted
                // 异常。。
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
                // 那什么时候就结束上面的自旋呢?一个是当前的线程被
                // signal了,那node就被transfer到sync queue了,while
                // 就不满足了。再一个就是线程中断了,在while循环体里给break掉了
            }
            // 跳出来后,紧接着去竞争锁,知道成功为止。&& 后面这个THROW_IE,标识
            // 要抛出异常,不是的话,就是REINTERRPUT,代表保证线程的中断标记不被
            // 重置即可。
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            // 这儿是在condition queue里有多个waiter的时候才起作用,主要用来将
            // CANCEL的结点从链队中剔除掉
            // 具体大家自己看吧。现在忽略这
            if (node.nextWaiter != null) // clean up if cancelled
                unlinkCancelledWaiters();
            // 这儿就是处理interruptMode中断标记字段的逻辑
            // 在reportxxx中,interruptMode为THROW_IE,则抛出
            // 异常,不是,则保证线程的中断field不被重置为“未中断”即可
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
        }
复制代码
  • consumer#signal

consumer 在调用 emptyCondition.signal 的时候,会影响到 emptyConditioncondition queue 中的等待线程,这里 具体指上面的provider#await方法。

public final void signal() {
            // 先判断下,lock锁是不是在调用signal方法的当前线程手里
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            // 取到condition queue里的第一个waiter node,这里也就是
            // consumer,因为它第一个await进入condition queue了
            Node first = firstWaiter;
            // 这里去进行了具体的signal操作,具体会做先把waiter node的waitStatus
            // 从CONDITION状态改为入Sync Queue的正常状态值0
            // 然后修改Sync Queue 的Head Tail等,让其入队成功
            // 最后再从其前驱结点的状态值上确保当前结点能够被唤起即可。
            // 这里是因为这个waitStatus值对后继结点的行为是有影响的,像SIGNAL指
            // 的是在结点释放后,要去唤醒后继结点
            // 
            if (first != null)
                doSignal(first);
        }
复制代码
  • consumer#unlock

unlock 具体调用的 AQSrelease() 方法

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

    // AQS.release
    public final boolean release(int arg) {
        // tryRelease,这里由NonFairSync实现,具体就是通过
        // CAS去修改state值,并判断是否成功释放锁
        if (tryRelease(arg)) {
            // 成功释放了,则在waitStatus 不是初始状态时,去唤醒后继,
            // 这个 != 0 来做判断的原因,就要综合所有情况,
            // 像FailSync NonFairSync / Exclusive / Share
            // 等所有情况来看这里的waitSTatus都会处于什么状态。
            // 全撸一遍的话,会发现这里的 != 0能够涵盖以上所有情况。
            Node h = head;
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }
复制代码
  • provider#unlock

这里就同理上面了。

原文  https://juejin.im/post/5c492b4ee51d456775678d25
正文到此结束
Loading...