首页 > 编程语言 >Java同步器之Condition源码分析

Java同步器之Condition源码分析

时间:2022-12-15 18:22:23浏览次数:57  
标签:node Java AQS 队列 源码 线程 条件 节点 Condition

一、概述

条件锁,就是指在获取锁之后发现当前业务场景自己无法处理,而需要等待某个条件的出现才可以继续处理时使用的一种锁。

比如,在阻塞队列中,当队列中没有元素的时候是无法弹出一个元素的,这时候就需要阻塞在条件notEmpty上,等待其它线程往里面放入一个元素后,唤醒这个条件notEmpty,当前线程才可以继续去做“弹出一个元素”的行为。

注意,这里的条件,必须是在获取锁之后去等待,对应到ReentrantLock的条件锁,就是获取锁之后才能调用condition.await()方法。

Java中,条件锁的实现都在AQSConditionObject类中,ConditionObject实现了Condition接口,所以ReentrantLock的条件锁是基于AQS实现的。

sychronized + Object.wait = Lock + Condition

二、案例

public class ReentrantLockTest {
    public static void main(String[] args) throws InterruptedException {
        // 声明一个重入锁
        ReentrantLock lock = new ReentrantLock();

        // 声明一个条件锁
        Condition condition = lock.newCondition();

        // 创建一个线程并执行,该线程就是当达到条件锁条件后,来执行后续的相关逻辑的。
        // 可以看作是一个消费者
        new Thread(() -> {
            try {
                // 获取锁
                lock.lock();  // 1
                try {
                    System.out.println("before await");  // 2
                    // 等待条件   
                    // 该线程执行到这里就会进入到阻塞状态,直到达到了条件后,
                    // 由其他线程执行signal()方法来告知该线程已经达到条件了,
                    // 该线程就会在这里被唤醒继续向下执行
                    condition.await();  // 3
                    System.out.println("after await");  // 10
                } finally {
                    // 释放锁
                    lock.unlock();  // 11
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        // 这里睡1000ms是为了让上面的线程先获取到锁
        Thread.sleep(1000);
        // main方法的当前线程来获取锁,该线程就是用来进行相关业务处理,进而达到条件锁条件的。
        // 可以看作是一个生产者
        lock.lock();  // 4
        try {
            // 这里睡2000ms代表这个线程执行业务需要的时间,
            // 当完成这里的2000ms之后就认为是符合条件锁条件了
            Thread.sleep(2000);  // 5
            System.out.println("before signal");  // 6
            // 通知条件已成立
            condition.signal();  // 7
            System.out.println("after signal");  // 8
        } finally {
            // 释放锁
            lock.unlock();  // 9
        }
    }
}

上面的代码很简单,一个线程等待条件,另一个线程通知条件已成立,后面的数字代表代码实际运行的顺序。

由上面的例子我们也可以看出,await()signal()方法都必须在获取锁之后释放锁之前使用;

三、源码分析

3.1 Condition接口

Condition是一个接口,从1.5的时候出现的,是用来替代Objectwaitnotify。所以显而易见,Conditionawaitsignal肯定效率更高、安全性更好。Condition是依赖于lock实现的。并且awaitsignal只能在lock的保护之内使用。

package java.util.concurrent.locks;

import java.util.concurrent.TimeUnit;
import java.util.Date;

public interface Condition {

    //导致当前线程等到发信号或 interrupted 。
    void await() throws InterruptedException;

    //使当前线程等待直到发出信号
    void awaitUninterruptibly();

    //使当前线程等待直到发出信号或中断,或指定的等待时间过去。 
    long awaitNanos(long nanosTimeout) throws InterruptedException;

    //使当前线程等待直到发出信号或中断,或指定的等待时间过去。 
    boolean await(long time, TimeUnit unit) throws InterruptedException;

    //使当前线程等待直到发出信号或中断,或者指定的最后期限过去。 
    boolean awaitUntil(Date deadline) throws InterruptedException;

    //唤醒一个等待线程。 
    void signal();

    //唤醒所有等待线程。 
    void signalAll();
}

3.2 内部类

ConditionObjectCondition的实现类。

//AbstractQueuedSynchronizer.ConditionObject
public class ConditionObject implements Condition, java.io.Serializable {

    /** First node of condition queue. */
    private transient Node firstWaiter;

    /** Last node of condition queue. */
    private transient Node lastWaiter;

    //...
}

可以看到条件锁中也维护了一个队列,为了和AQS的队列区分,我这里称为条件队列,firstWaiter是队列的头节点,lastWaiter是队列的尾节点。

ConditionObject中的条件队列和AQS中的同步队列使用的是相同的节点类型Node,但是两个队列还是有一些不同的,在后续会详细讲解。

3.3 lock.newCondition()方法

新建一个条件锁。

  • ReentrantLock.newCondition()
  • ReentrantLock.Sync.newCondition()
  • AbstractQueuedSynchronizer.ConditionObject.ConditionObject()
public class ReentrantLock implements Lock, java.io.Serializable {

    private final Sync sync;
    
    // 创建条件锁
    public Condition newCondition() {
        return sync.newCondition();
    }
    
    /**
     * 抽象内部类
     */
    abstract static class Sync extends AbstractQueuedSynchronizer {
        // 条件锁
        final ConditionObject newCondition() {
            return new ConditionObject();
        }
    }
}
 
// AbstractQueuedSynchronizer.ConditionObject.ConditionObject()
public ConditionObject() { }

新建一个条件锁最后就是调用的AQS中的ConditionObject类来实例化条件锁。

3.4 condition.await()方法

condition.await()方法,表明现在要等待条件的出现。

// AbstractQueuedSynchronizer.ConditionObject.await()
public final void await() throws InterruptedException {
    // 如果线程中断了,抛出异常
    if (Thread.interrupted())
        throw new InterruptedException();
    // 添加节点到Condition的队列中,并返回该节点
    Node node = addConditionWaiter();
    // 完全释放当前线程获取的锁
    // 因为锁是可重入的,所以这里要把获取的锁全部释放
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    // 是否在同步队列中
    while (!isOnSyncQueue(node)) {
        // 阻塞当前线程
        LockSupport.park(this);
        
        // 上面部分是调用await()时释放自己占有的锁,并阻塞自己等待条件的出现

        // *************************分界线*************************  //

        // 下面部分是条件已经出现,尝试去获取锁
        
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    
    // 尝试获取锁,注意第二个参数,这是上一章分析过的方法
    // 如果没获取到会再次阻塞(这个方法这里就不贴出来了,有兴趣的翻翻上一章的内容)
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    // 清除取消的节点
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    // 线程中断相关
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

// AbstractQueuedSynchronizer.ConditionObject.addConditionWaiter
private Node addConditionWaiter() {
    Node t = lastWaiter;
    // 如果条件队列的尾节点已取消,从头节点开始清除所有已取消的节点
    if (t != null && t.waitStatus != Node.CONDITION) {
        unlinkCancelledWaiters();
        // 重新获取尾节点
        t = lastWaiter;
    }
    // 新建一个节点,它的等待状态是CONDITION
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    // 如果尾节点为空,则把新节点赋值给头节点(相当于初始化队列)
    // 否则把新节点赋值给尾节点的nextWaiter指针
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    // 尾节点指向新节点
    lastWaiter = node;
    // 返回新节点
    return node;
}

// AbstractQueuedSynchronizer.fullyRelease
final int fullyRelease(Node node) {
    boolean failed = true;
    try {
        // 获取状态变量的值,重复获取锁,这个值会一直累加
        // 所以这个值也代表着获取锁的次数
        int savedState = getState();
        // 一次性释放所有获得的锁
        if (release(savedState)) {
            failed = false;
            // 返回获取锁的次数
            return savedState;
        } else {
            throw new IllegalMonitorStateException();
        }
    } finally {
        if (failed)
            node.waitStatus = Node.CANCELLED;
    }
}

// AbstractQueuedSynchronizer.isOnSyncQueue
final boolean isOnSyncQueue(Node node) {
    // 如果等待状态是CONDITION,或者前一个指针为空,返回false
    // 说明还没有移到AQS的队列中
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        return false;
    // 如果next指针有值,说明已经移到AQS的队列中了
    if (node.next != null) // If has successor, it must be on queue
        return true;
    // 从AQS的尾节点开始往前寻找看是否可以找到当前节点,找到了也说明已经在AQS的队列中了
    return findNodeFromTail(node);
}

这里有几个难理解的点:

  1. Condition的队列和AQS的队列不完全一样;
  • AQS的队列头节点是不存在任何值的,是一个虚节点;
  • Condition的队列头节点是存储着实实在在的元素值的,是真实节点。
  1. 各种等待状态(waitStatus)的变化;
  • 首先,在条件队列中,新建节点的初始等待状态是CONDITION(-2)
  • 其次,移到AQS的队列中时等待状态会更改为0(AQS队列节点的初始等待状态为0);
  • 然后,在AQS的队列中如果需要阻塞,会把它上一个节点的等待状态设置为SIGNAL(-1)
  • 最后,不管在Condition队列还是AQS队列中,已取消的节点的等待状态都会设置为CANCELLED(1)
  • 另外,后面我们在共享锁的时候还会讲到另外一种等待状态叫PROPAGATE(-3)
  1. 相似的名称;
  • AQS中下一个节点是next,上一个节点是prev
  • Condition中下一个节点是nextWaiter,没有上一个节点。

总结一下await()方法的大致流程:

  1. 新建一个节点加入到条件队列中去;
  2. 完全释放当前线程占有的锁;
  3. 阻塞当前线程,并等待条件的出现;
  4. 条件已出现(此时节点已经移到AQS的队列中),尝试获取锁;

也就是说await()方法内部其实是先释放锁->等待条件->再次获取锁的过程。

3.5 condition.signal()方法

condition.signal()方法通知条件已经出现。

// AbstractQueuedSynchronizer.ConditionObject.signal
public final void signal() {
    // 如果不是当前线程占有着锁,调用这个方法抛出异常
    // 说明signal()也要在获取锁之后执行
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    // 条件队列的头节点
    Node first = firstWaiter;
    // 如果有等待条件的节点,则通知它条件已成立
    if (first != null)
        doSignal(first);
}

// AbstractQueuedSynchronizer.ConditionObject.doSignal
private void doSignal(Node first) {
    do {
        // 移到条件队列的头节点往后一位
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        // 相当于把头节点从队列中出队
        first.nextWaiter = null;
        // 转移节点到AQS队列中
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
}

// AbstractQueuedSynchronizer.transferForSignal
final boolean transferForSignal(Node node) {
    // 把节点的状态更改为0,也就是说即将移到AQS队列中
    // 如果失败了,说明节点已经被改成取消状态了
    // 返回false,通过上面的循环可知会寻找下一个可用节点
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;

    // 调用AQS的入队方法把节点移到AQS的队列中
    // 注意,这里enq()的返回值是node的上一个节点,也就是旧尾节点
    Node p = enq(node);
    // 上一个节点的等待状态
    int ws = p.waitStatus;
    // 如果上一个节点已取消了,或者更新状态为SIGNAL失败(也是说明上一个节点已经取消了)
    // 则直接唤醒当前节点对应的线程
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);
    // 如果更新上一个节点的等待状态为SIGNAL成功了
    // 则返回true,这时上面的循环不成立了,退出循环,也就是只通知了一个节点
    // 此时当前节点还是阻塞状态
    // 也就是说调用signal()的时候并不会真正唤醒一个节点
    // 只是把节点从条件队列移到AQS队列中
    return true;
}

signal()方法的大致流程为:

  1. 从条件队列的头节点开始寻找一个非取消状态的节点;
  2. 把它从条件队列移到AQS队列;
  3. 且只移动一个节点;

注意,这里调用signal()方法后并不会真正唤醒一个节点,那么,唤醒一个节点是在啥时候呢?

我们可以再回去看一下本文最开始的使用案例,在其他线程调用signal()方法后,该线程最终会执行lock.unlock()方法,此时才会真正唤醒一个正在同步队列中的节点,唤醒的这个节点如果曾经是条件节点被转移到条件队列中的话,就会继续执行await()方法“分界线”下面的代码。也就是说,在调用signal()方法的线程调用unlock()方法才是真正唤醒阻塞在条件上的节点(此时节点已经在AQS队列中);被唤醒之后,被唤醒的节点会再次尝试获取锁,后面的逻辑与lock()的逻辑基本一致了。

3.6 condition.signalAll()方法

signalAllsignal类似,区别是signalAll会将所有节点加入同步队列,除了doSignalAll方法以外,其他的方法都是一样的:

private void doSignalAll(Node first) {
    // 将条件队列的头节点和尾节点都置为null
    lastWaiter = firstWaiter = null;
    // 遍历条件队列,依次唤醒所有线程
    do {
        Node next = first.nextWaiter;
        first.nextWaiter = null;
        transferForSignal(first);
        first = next;
    } while (first != null);
}

四、总结

  1. 条件锁是指为了等待某个条件出现而使用的一种锁;
  2. 条件锁比较经典的使用场景就是队列为空时阻塞在条件notEmpty上;
  3. ReentrantLock中的条件锁是通过AQS的ConditionObject内部类实现的;
  4. await()和signal()方法都必须在获取锁之后释放锁之前使用;
  5. await()方法会新建一个节点放到条件队列中,接着完全释放锁,然后阻塞当前线程并等待条件的出现;
  6. signal()方法会寻找条件队列中第一个可用节点移到AQS队列中;
  7. 在调用signal()方法的线程调用unlock()方法才真正唤醒阻塞在条件上的节点(此时节点已经在AQS队列中);
  8. 之后该节点会再次尝试获取锁,后面的逻辑与lock()的逻辑基本一致了。

参考文章

标签:node,Java,AQS,队列,源码,线程,条件,节点,Condition
From: https://www.cnblogs.com/ciel717/p/16985782.html

相关文章

  • maven fmpp+javacc 集成使用简单说明
    dremio以及apachecalcite使用到fmpp+javacc进行代码生成处理,以下是一个简单的集成测试fmpp的作用fmpp实际上是包装了freemarker,提供了cli以及javaapi可以方......
  • java maven工程打出可执行jar包
    javamaven工程打出可执行jar包,在pom.xml中添加如下配置即可<build><plugins><!--设置编译版本--><plugin><groupId>org.apache.maven.......
  • Java-快速入门、IDEA开发工具的使用1-笔记
    1.Java的概述Java是sun公司1995推出的,2009年被oracle收购Java的爸爸:詹姆斯.高斯林Java是一门高级编程语言:语言风格接近人类的自然语言,写程序简单易懂Java的流行度很高,商业占......
  • Java:SpringBoot使用EasyExcel实现Excel文件的导出下载和上传导入功能
    SpringBoot使用EasyExcel实现Excel文件的导出下载和上传导入功能文件目录$tree-Itarget.├──README.md├──pom.xml└──src└──main├─......
  • Java进阶——线程基础
    线程基础1、线程介绍1.1、线程相关概念程序(program):是完成特定任务,用某种语言编写的一组的集合(例如:java代码就是一个程序)进程:是指运行中的程序(比如:我现在使用的Typora)......
  • Java进阶——IO流
    IO流一、文件1.1、文件流文件在程序中是以流的形式来操作的流:数据在数据源(文件)和程序(内存)之间经历的路径输入流:数据从数据源(文件)到程序(内存)的路径输入流:数据从程序(内存)到数......
  • JavaconfigApplicationTests
    packagecom.example.demo;importcom.example.demo.config.JavaConfig;importcom.example.demo.model.Printer;importorg.junit.Test;importorg.junit.runner.Ru......
  • java DefaultMutableTreeNode 树形结构
    目录1.Tree的概念11.1.treeNode接口,mutabletreenode接口11.2.10-4:以TreeModel构造JTree.12.功能要求22.1.树形结构节点的查找与定位23.节点......
  • java+Selenium(一)八种元素获取方式
    官方文档:https://www.selenium.dev/documentationchromedriver下载地址:注意:需要下载与浏览器匹配版本的http://chromedriver.storage.googleapis.com/index.htmlhttp://......
  • Java数组(3)三种初始化及内存分析
           ......