Condition的使用与源码进阶

 

一、Condition作用是什么

  主要功能和Object中的wait、notify功能相对应，使某个线程 在某种情况下等待和唤醒的功能。

 

 

二、使用实例

1）实例1，单一生产者和消费者使用，并且只用一个condition对象控制生产者和消费者

注意：单一condition对象，生产者和消费者都在一个队列中排队，那么这个时候下一个被唤醒的可能是消费者也可能是生产者，那么如果生产者需要唤醒消费者但是被唤醒的确实生产者，则需要生产者继续阻塞排队（使用循环判断模式）

 

 

package com.test.lock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.LinkedList;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ConditionTest4 {

    static Logger logger = LoggerFactory.getLogger(ConditionTest4.class ) ;

    public static int getRandom(){
        Random random = new Random() ;
        int rid = random.nextInt(6);
        int val =  rid+2;
        logger.info(String.format("生成随机数：%d", val ));
        return  val;
    }


    //睡眠打印日志更易于观察
    public static void waitN(Integer n ){
        try {
            TimeUnit.SECONDS.sleep(n);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return   ;
    }

    public static void main(String[] args) throws InterruptedException {

        Lock lock = new ReentrantLock(true) ;
        Condition condition = lock.newCondition();

        //队列允许的最大值
        Integer count=10;
       //数据等待的队列
        LinkedList<Integer> list = new LinkedList<>() ;


        Thread producter = new Thread(()->{
            while (true){//保持生产者一直可以生产
                lock.lock();
                try {
                    while(list.size()==count){//保证 condition被叫醒以后，集合中不是满的
                        try {
                            logger.info("1111111111生产者进入等待队列");
                            waitN(2);//睡眠，使测试更容易被观看结果
                            condition.await();
                            logger.info("111111111生产者被唤醒");
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    logger.info("111111111111生产者可以生产了");
                    waitN(2);
                    list.add( getRandom());
                    condition.signal();
                }finally {
                    lock.unlock();
                }
            }
        },"producter") ;
        producter.start();

        Thread consumer = new Thread(()->{
            while (true){
                lock.lock();

                while(list.size()==0){//每次唤醒判断如果是空的则继续排队
                    try {
                        logger.info("22222222消费者进入等待队列");
                        waitN(5);
                        condition.await();
                        logger.info("222222222消费者被唤醒");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                logger.info("2222222222消费者可以消费了");
                waitN(2);
                int val =list.pop();
                logger.info(String.format("2222222消费者可以消费了 %d ", val ));
                condition.signal();

                //解锁一般放到finnaly中
                lock.unlock();
            }
        },"consumer") ;
        consumer.start();
    }
}

2）实例2，多个生产者和多个消费者使用，生产者和消费者分别使用自己的condition对象

 当生产者的线程大于消费者的线程，则最终队列会在 满的地方徘徊，

package com.test.lock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.LinkedList;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ConditionTest5 {

    static Logger logger = LoggerFactory.getLogger(ConditionTest5.class ) ;

    public static int getRandom(){
        Random random = new Random() ;
        int rid = random.nextInt(6);
        int val =  rid+2;
        logger.info(String.format("生成随机数：%d", val ));
        return  val;
    }

    //睡眠打印日志更易于观察
    public static void waitN(Integer n ){
        try {
            TimeUnit.SECONDS.sleep(n);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return   ;
    }

    public static void main(String[] args) throws InterruptedException {
        //true 公平锁，这样生产者和消费者才能交替执行
        Lock lock = new ReentrantLock(true) ;
        //2个condition对象，一个在生产者中等待，一个在消费者中等待
        Condition produCondition = lock.newCondition();
        Condition consuCondition = lock.newCondition();

        Integer count=10;
        LinkedList<Integer> list = new LinkedList<>() ;

        for (int i = 0; i < 4 ; i++) {
            Thread producter = new Thread(()->{
                while (true){//保持生产者一直可以生产
                    lock.lock();

                    while(list.size()==count){//保证 condition被叫醒以后，集合中不是满的
                        try {
                            logger.info("1111111111生产者进入等待队列");
                            waitN(2);
                            produCondition.await();
                            logger.info("111111111生产者被唤醒");
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    logger.info("111111111111生产者可以生产了, {}" , list.toString());
                    waitN(2);
                    list.add( getRandom());
                    consuCondition.signal();

                    lock.unlock();
                }
            },"producter"+i) ;
            producter.start();
        }

        for (int i = 0; i < 2; i++) {
            Thread consumer = new Thread(()->{
                while (true){
                    lock.lock();
                    while(list.size()==0){
                        try {
                            logger.info("22222222消费者进入等待队列");
                            waitN(5);
                            consuCondition.await();
                            logger.info("222222222消费者被唤醒");
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    logger.info("2222222222消费者可以消费了");
                    waitN(2);
                    int val =list.pop();
                    logger.info(String.format("2222222消费者可以消费了 %d ,  %s ", val ,  list.toString()  ));
                    produCondition.signal();
                    lock.unlock();
                }
            },"consumer"+i ) ;
            consumer.start();
        }
    }
}

三、源码主要方法解读

 

内部类：https://www.cnblogs.com/qdhxhz/p/11368870.html

Condition这个类是AQS的内部类，通过lock进行实例化的，当我们使用condition对象内部方法的时候，是可以使用lock对象的属性的。

 

1）await方法

 

public final void await() throws InterruptedException {

    if (Thread.interrupted())

        throw new InterruptedException();

    //当前线程进入等待队列

    Node node = addConditionWaiter();

    long savedState = fullyRelease(node);

    int interruptMode = 0;

   //判断节点是否在lock的等待队列中，如果已经不在了则这个线程等待

    while (!isOnSyncQueue(node)) {

        //如果没在lock的等待队列中，则当前线程睡眠

        LockSupport.park(this);

        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);

}

 2）当前线程入队方法

  

private Node addConditionWaiter() {
    Node t = lastWaiter;
    // If lastWaiter is cancelled, clean out.
    //如果队列中的状态不是CONDITION，则清除掉这个节点
    if (t != null && t.waitStatus != Node.CONDITION) {
        unlinkCancelledWaiters();
        t = lastWaiter;
    }
    // 形成一个节点
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
   
    if (t == null)
        //如果第一个线程进入等待队列，则当前节点为第一个节点 
        firstWaiter = node;
    else
        //如果不是第一个进入等待，则最后一个节点的next指向当前节点
        t.nextWaiter = node;
    //当前线程的节点变成最后的节点
    lastWaiter = node;
   //返回当前线程的节点
    return node;
}

3）线程等待了，则会释放当前线程持有的锁

final int fullyRelease(Node node) {
    boolean failed = true;
    try {
        int savedState = getState();
        //调用外部类对象lock的方法进行锁释放
        if (release(savedState)) {
            failed = false;
            return savedState;
        } else {
            throw new IllegalMonitorStateException();
        }
    } finally {
        //如果释放锁失败则当前线程的节点状态为取消 
        if (failed)
            node.waitStatus = Node.CANCELLED;
    }
}

 4） 当前节点是否在lock的等待队列中

final boolean isOnSyncQueue(Node node) {
    // 当前节点状态为CONDITION ，并且是head节点（节点为head节点： node.prev == null）
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        return false;
    if (node.next != null) // If has successor, it must be on queue
        return true;
    /*
     * node.prev can be non-null, but not yet on queue because
     * the CAS to place it on queue can fail. So we have to
     * traverse from tail to make sure it actually made it.  It
     * will always be near the tail in calls to this method, and
     * unless the CAS failed (which is unlikely), it will be
     * there, so we hardly ever traverse much.
     */
    return findNodeFromTail(node);
}

5）从lock的队列的尾部 向前查找是否可以找到这个节点，找到则返回true，找到head节点还没找到

  

private boolean findNodeFromTail(Node node) {
    Node t = tail;
    for (;;) {
        if (t == node)
            return true;
         //找到head节点还是没有，则返回false，head节点的prev是null
        if (t == null)
            return false;
        t = t.prev;
    }
}

 6）唤醒睡眠的线程

  

public final void signal() {
    //必须是当前获取锁的线程来执行
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
        //唤醒队列中第一个线程
        doSignal(first);
}

 7）唤醒第一个线程，清除第一个node对应的next信息

    

private void doSignal(Node first) {
    do {
        //如果只有一个节点，则清除lastWaiter ，如果first唤醒失败，则把first的next 赋值给firstWaiter ，在此执行唤醒
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        //清除nextWaiter信息
        first.nextWaiter = null;
    } while (!transferForSignal(first) && //去唤醒第一个节点 
             (first = firstWaiter) != null);  // 判断first!= null 
}

 8  唤醒第一个节点去获取锁

  

final boolean transferForSignal(Node node) {
    /*
     * If cannot change waitStatus, the node has been cancelled.
     */
    //设置节点状态 从 CONDITION 变为 0
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        //如果失败，则直接返回
        return false;


    /*
     * Splice onto queue and try to set waitStatus of predecessor to
     * indicate that thread is (probably) waiting. If cancelled or
     * attempt to set waitStatus fails, wake up to resync (in which
     * case the waitStatus can be transiently and harmlessly wrong).
     */
    //节点放到lock队列中去抢锁
    Node p = enq(node);
    int ws = p.waitStatus;
    if (ws > 0  //节点状态为1 ，被取消了 
          || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) //设置线程节点状态为SIGNAL，并判断成功与失败
        //线程在上一个步骤已经放到了lock队列中，这里只是设置一下node状态，实际上在lock.unlock步骤会从队列中唤醒的
        LockSupport.unpark(node.thread);
    return true;
}