java高并发系列 - 第13天:JUC中的Condition对象

时间  2019-11-06
标签 java 并发 系列 13天 juc condition 对象 栏目 Java 繁體版
原文   原文链接

本文目标：java

  1. synchronized中实现线程等待和唤醒
  2. Condition简介及经常使用方法介绍及相关示例
  3. 使用Condition实现生产者消费者
  4. 使用Condition实现同步阻塞队列

Object对象中的wait()，notify()方法，用于线程等待和唤醒等待中的线程，你们应该比较熟悉，想再次了解的朋友能够移步到线程的基本操做微信

synchronized中等待和唤醒线程示例

package com.itsoku.chat09;

import java.util.concurrent.TimeUnit;

/**
 * 微信公众号：javacode2018，获取年薪50万课程
 */
public class Demo1 {
    static Object lock = new Object();

    public static class T1 extends Thread {
        @Override
        public void run() {
            System.out.println(System.currentTimeMillis() + "," + this.getName() + "准备获取锁!");
            synchronized (lock) {
                System.out.println(System.currentTimeMillis() + "," + this.getName() + "获取锁成功!");
                try {
                    lock.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            System.out.println(System.currentTimeMillis() + "," + this.getName() + "释放锁成功!");
        }
    }

    public static class T2 extends Thread {
        @Override
        public void run() {
            System.out.println(System.currentTimeMillis() + "," + this.getName() + "准备获取锁!");
            synchronized (lock) {
                System.out.println(System.currentTimeMillis() + "," + this.getName() + "获取锁成功!");
                lock.notify();
                System.out.println(System.currentTimeMillis() + "," + this.getName() + " notify!");
                try {
                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(System.currentTimeMillis() + "," + this.getName() + "准备释放锁!");
            }
            System.out.println(System.currentTimeMillis() + "," + this.getName() + "释放锁成功!");
        }
    }

    public static void main(String[] args) throws InterruptedException {
        T1 t1 = new T1();
        t1.setName("t1");
        t1.start();
        TimeUnit.SECONDS.sleep(5);
        T2 t2 = new T2();
        t2.setName("t2");
        t2.start();
    }
}

输出：并发

1：1563530109234,t1准备获取锁!
2：1563530109234,t1获取锁成功!
3：1563530114236,t2准备获取锁!
4：1563530114236,t2获取锁成功!
5：1563530114236,t2 notify!
6：1563530119237,t2准备释放锁!
7：1563530119237,t2释放锁成功!
8：1563530119237,t1释放锁成功!

代码结合输出的结果咱们分析一下：ide

  1. 线程t1先获取锁，而后调用了wait()方法将线程置为等待状态，而后会释放lock的锁
  2. 主线程等待5秒以后，启动线程t2，t2获取到了锁，结果中一、3行时间相差5秒左右
  3. t2调用lock.notify()方法，准备将等待在lock上的线程t1唤醒，notify()方法以后又休眠了5秒，看一下输出的五、8可知，notify()方法以后，t1并不能当即被唤醒，须要等到t2将synchronized块执行完毕，释放锁以后，t1才被唤醒
  4. wait()方法和notify()方法必须放在同步块内调用（synchronized块内），不然会报错

Condition使用简介

在了解Condition以前，须要先了解一下重入锁ReentrantLock，能够移步到：JUC中的ReentranLock。高并发

任何一个java对象都自然继承于Object类，在线程间实现通讯的每每会应用到Object的几个方法，好比wait()、wait(long timeout)、wait(long timeout, int nanos)与notify()、notifyAll()几个方法实现等待/通知机制，一样的， 在java Lock体系下依然会有一样的方法实现等待/通知机制。this

从总体上来看Object的wait和notify/notify是与对象监视器配合完成线程间的等待/通知机制，而Condition与Lock配合完成等待通知机制，前者是java底层级别的，后者是语言级别的，具备更高的可控制性和扩展性。二者除了在使用方式上不一样外，在功能特性上仍是有不少的不一样：线程

  1. Condition可以支持不响应中断，而经过使用Object方式不支持
  2. Condition可以支持多个等待队列（new 多个Condition对象），而Object方式只能支持一个
  3. Condition可以支持超时时间的设置，而Object不支持

Condition由ReentrantLock对象建立，而且能够同时建立多个，Condition接口在使用前必须先调用ReentrantLock的lock()方法得到锁，以后调用Condition接口的await()将释放锁，而且在该Condition上等待，直到有其余线程调用Condition的signal()方法唤醒线程，使用方式和wait()、notify()相似。3d

示例代码：code

package com.itsoku.chat09;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 微信公众号：javacode2018，获取年薪50万课程
 */
public class Demo2 {
    static ReentrantLock lock = new ReentrantLock();
    static Condition condition = lock.newCondition();

    public static class T1 extends Thread {
        @Override
        public void run() {
            System.out.println(System.currentTimeMillis() + "," + this.getName() + "准备获取锁!");
            lock.lock();
            try {
                System.out.println(System.currentTimeMillis() + "," + this.getName() + "获取锁成功!");
                condition.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
            System.out.println(System.currentTimeMillis() + "," + this.getName() + "释放锁成功!");
        }
    }

    public static class T2 extends Thread {
        @Override
        public void run() {
            System.out.println(System.currentTimeMillis() + "," + this.getName() + "准备获取锁!");
            lock.lock();
            try {
                System.out.println(System.currentTimeMillis() + "," + this.getName() + "获取锁成功!");
                condition.signal();
                System.out.println(System.currentTimeMillis() + "," + this.getName() + " signal!");
                try {
                    TimeUnit.SECONDS.sleep(5);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println(System.currentTimeMillis() + "," + this.getName() + "准备释放锁!");
            } finally {
                lock.unlock();
            }
            System.out.println(System.currentTimeMillis() + "," + this.getName() + "释放锁成功!");
        }
    }

    public static void main(String[] args) throws InterruptedException {
        T1 t1 = new T1();
        t1.setName("t1");
        t1.start();
        TimeUnit.SECONDS.sleep(5);
        T2 t2 = new T2();
        t2.setName("t2");
        t2.start();
    }
}

输出：对象

1563532185827,t1准备获取锁!
1563532185827,t1获取锁成功!
1563532190829,t2准备获取锁!
1563532190829,t2获取锁成功!
1563532190829,t2 signal!
1563532195829,t2准备释放锁!
1563532195829,t2释放锁成功!
1563532195829,t1释放锁成功!

输出的结果和使用synchronized关键字的实例相似。

Condition.await()方法和Object.wait()方法相似，当使用Condition.await()方法时，须要先获取Condition对象关联的ReentrantLock的锁，在Condition.await()方法被调用时，当前线程会释放这个锁，而且当前线程会进行等待（处于阻塞状态）。在signal()方法被调用后，系统会从Condition对象的等待队列中唤醒一个线程，一旦线程被唤醒，被唤醒的线程会尝试从新获取锁，一旦获取成功，就能够继续执行了。所以，在signal被调用后，通常须要释放相关的锁，让给其余被唤醒的线程，让他能够继续执行。

Condition经常使用方法

Condition接口提供的经常使用方法有：

和Object中wait相似的方法

  1. void await() throws InterruptedException:当前线程进入等待状态，若是其余线程调用condition的signal或者signalAll方法而且当前线程获取Lock从await方法返回，若是在等待状态中被中断会抛出被中断异常；
  2. long awaitNanos(long nanosTimeout)：当前线程进入等待状态直到被通知，中断或者超时；
  3. boolean await(long time, TimeUnit unit) throws InterruptedException：同第二种，支持自定义时间单位，false：表示方法超时以后自动返回的，true：表示等待还未超时时，await方法就返回了（超时以前，被其余线程唤醒了）
  4. boolean awaitUntil(Date deadline) throws InterruptedException：当前线程进入等待状态直到被通知，中断或者到了某个时间
  5. void awaitUninterruptibly();：当前线程进入等待状态，不会响应线程中断操做，只能经过唤醒的方式让线程继续

和Object的notify/notifyAll相似的方法

  1. void signal()：唤醒一个等待在condition上的线程，将该线程从等待队列中转移到同步队列中，若是在同步队列中可以竞争到Lock则能够从等待方法中返回。
  2. void signalAll()：与1的区别在于可以唤醒全部等待在condition上的线程

Condition.await()过程当中被打断

package com.itsoku.chat09;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 微信公众号：javacode2018，获取年薪50万课程
 */
public class Demo4 {
    static ReentrantLock lock = new ReentrantLock();
    static Condition condition = lock.newCondition();

    public static class T1 extends Thread {
        @Override
        public void run() {
            lock.lock();
            try {
                condition.await();
            } catch (InterruptedException e) {
                System.out.println("中断标志：" + this.isInterrupted());
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        T1 t1 = new T1();
        t1.setName("t1");
        t1.start();
        TimeUnit.SECONDS.sleep(2);
        //给t1线程发送中断信号
        System.out.println("一、t1中断标志：" + t1.isInterrupted());
        t1.interrupt();
        System.out.println("二、t1中断标志：" + t1.isInterrupted());
    }
}

输出：

一、t1中断标志：false
二、t1中断标志：true
中断标志：false
java.lang.InterruptedException
    at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.reportInterruptAfterWait(AbstractQueuedSynchronizer.java:2014)
    at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2048)
    at com.itsoku.chat09.Demo4$T1.run(Demo4.java:19)

调用condition.await()以后，线程进入阻塞中，调用t1.interrupt()，给t1线程发送中断信号，await()方法内部会检测到线程中断信号，而后触发InterruptedException异常，线程中断标志被清除。从输出结果中能够看出，线程t1中断标志的变换过程：false->true->false

await(long time, TimeUnit unit)超时以后自动返回

package com.itsoku.chat09;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 微信公众号：javacode2018，获取年薪50万课程
 */
public class Demo5 {
    static ReentrantLock lock = new ReentrantLock();
    static Condition condition = lock.newCondition();

    public static class T1 extends Thread {
        @Override
        public void run() {
            lock.lock();
            try {
                System.out.println(System.currentTimeMillis() + "," + this.getName() + ",start");
                boolean r = condition.await(2, TimeUnit.SECONDS);
                System.out.println(r);
                System.out.println(System.currentTimeMillis() + "," + this.getName() + ",end");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        T1 t1 = new T1();
        t1.setName("t1");
        t1.start();
    }
}

输出：

1563541624082,t1,start
false
1563541626085,t1,end

t1线程等待2秒以后，自动返回继续执行，最后await方法返回false，await返回false表示超时以后自动返回

await(long time, TimeUnit unit)超时以前被唤醒

package com.itsoku.chat09;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 微信公众号：javacode2018，获取年薪50万课程
 */
public class Demo6 {
    static ReentrantLock lock = new ReentrantLock();
    static Condition condition = lock.newCondition();

    public static class T1 extends Thread {
        @Override
        public void run() {
            lock.lock();
            try {
                System.out.println(System.currentTimeMillis() + "," + this.getName() + ",start");
                boolean r = condition.await(5, TimeUnit.SECONDS);
                System.out.println(r);
                System.out.println(System.currentTimeMillis() + "," + this.getName() + ",end");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        T1 t1 = new T1();
        t1.setName("t1");
        t1.start();
        //休眠1秒以后，唤醒t1线程
        TimeUnit.SECONDS.sleep(1);
        lock.lock();
        try {
            condition.signal();
        } finally {
            lock.unlock();
        }
    }
}

输出：

1563542046046,t1,start
true
1563542047048,t1,end

t1线程中调用condition.await(5, TimeUnit.SECONDS);方法会释放锁，等待5秒，主线程休眠1秒，而后获取锁，以后调用signal()方法唤醒t1，输出结果中发现await后过了1秒（一、3行输出结果的时间差），await方法就返回了，而且返回值是true。true表示await方法超时以前被其余线程唤醒了。

long awaitNanos(long nanosTimeout)超时返回

package com.itsoku.chat09;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 微信公众号：javacode2018，获取年薪50万课程
 */
public class Demo7 {
    static ReentrantLock lock = new ReentrantLock();
    static Condition condition = lock.newCondition();

    public static class T1 extends Thread {
        @Override
        public void run() {
            lock.lock();
            try {
                System.out.println(System.currentTimeMillis() + "," + this.getName() + ",start");
                long r = condition.awaitNanos(TimeUnit.SECONDS.toNanos(5));
                System.out.println(r);
                System.out.println(System.currentTimeMillis() + "," + this.getName() + ",end");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        T1 t1 = new T1();
        t1.setName("t1");
        t1.start();
    }
}

输出：

1563542547302,t1,start
-258200
1563542552304,t1,end

awaitNanos参数为纳秒，能够调用TimeUnit中的一些方法将时间转换为纳秒。

t1调用await方法等待5秒超时返回，返回结果为负数，表示超时以后返回的。

waitNanos(long nanosTimeout)超时以前被唤醒

package com.itsoku.chat09;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 微信公众号：javacode2018，获取年薪50万课程
 */
public class Demo8 {
    static ReentrantLock lock = new ReentrantLock();
    static Condition condition = lock.newCondition();

    public static class T1 extends Thread {
        @Override
        public void run() {
            lock.lock();
            try {
                System.out.println(System.currentTimeMillis() + "," + this.getName() + ",start");
                long r = condition.awaitNanos(TimeUnit.SECONDS.toNanos(5));
                System.out.println(r);
                System.out.println(System.currentTimeMillis() + "," + this.getName() + ",end");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        T1 t1 = new T1();
        t1.setName("t1");
        t1.start();
        //休眠1秒以后，唤醒t1线程
        TimeUnit.SECONDS.sleep(1);
        lock.lock();
        try {
            condition.signal();
        } finally {
            lock.unlock();
        }
    }
}

输出：

1563542915991,t1,start
3999988500
1563542916992,t1,end

t1中调用await休眠5秒，主线程休眠1秒以后，调用signal()唤醒线程t1，await方法返回正数，表示返回时距离超时时间还有多久，将近4秒，返回正数表示，线程在超时以前被唤醒了。

其余几个有参的await方法和无参的await方法同样，线程调用interrupt()方法时，这些方法都会触发InterruptedException异常，而且线程的中断标志会被清除。

同一个锁支持建立多个Condition

使用两个Condition来实现一个阻塞队列的例子：

package com.itsoku.chat09;

import java.util.LinkedList;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 微信公众号：javacode2018，获取年薪50万课程
 */
public class BlockingQueueDemo<E> {
    int size;//阻塞队列最大容量

    ReentrantLock lock = new ReentrantLock();

    LinkedList<E> list = new LinkedList<>();//队列底层实现

    Condition notFull = lock.newCondition();//队列满时的等待条件
    Condition notEmpty = lock.newCondition();//队列空时的等待条件

    public BlockingQueueDemo(int size) {
        this.size = size;
    }

    public void enqueue(E e) throws InterruptedException {
        lock.lock();
        try {
            while (list.size() == size)//队列已满,在notFull条件上等待
                notFull.await();
            list.add(e);//入队:加入链表末尾
            System.out.println("入队：" + e);
            notEmpty.signal(); //通知在notEmpty条件上等待的线程
        } finally {
            lock.unlock();
        }
    }

    public E dequeue() throws InterruptedException {
        E e;
        lock.lock();
        try {
            while (list.size() == 0)//队列为空,在notEmpty条件上等待
                notEmpty.await();
            e = list.removeFirst();//出队:移除链表首元素
            System.out.println("出队：" + e);
            notFull.signal();//通知在notFull条件上等待的线程
            return e;
        } finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) {
        BlockingQueueDemo<Integer> queue = new BlockingQueueDemo<>(2);
        for (int i = 0; i < 10; i++) {
            int data = i;
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        queue.enqueue(data);
                    } catch (InterruptedException e) {

                    }
                }
            }).start();
        }
        for (int i = 0; i < 10; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        Integer data = queue.dequeue();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        }
    }
}

代码很是容易理解，建立了一个阻塞队列，大小为3，队列满的时候，会被阻塞，等待其余线程去消费，队列中的元素被消费以后，会唤醒生产者，生产数据进入队列。上面代码将队列大小置为1，能够实现同步阻塞队列，生产1个元素以后，生产者会被阻塞，待消费者消费队列中的元素以后，生产者才能继续工做。

Object的监视器方法与Condition接口的对比

对比项 Object 监视器方法 Condition
前置条件 获取对象的锁 调用Lock.lock获取锁，调用Lock.newCondition()获取Condition对象
调用方式 直接调用，如：object.wait() 直接调用，如：condition.await()
等待队列个数 一个 多个，使用多个condition实现
当前线程释放锁并进入等待状态 支持 支持
当前线程释放锁进入等待状态中不响应中断 不支持 支持
当前线程释放锁并进入超时等待状态 支持 支持
当前线程释放锁并进入等待状态到未来某个时间 不支持 支持
唤醒等待队列中的一个线程 支持 支持
唤醒等待队列中的所有线程 支持 支持

总结

  1. 使用condition的步骤：建立condition对象，获取锁，而后调用condition的方法
  2. 一个ReentrantLock支持床多个condition对象
  3. void await() throws InterruptedException;方法会释放锁，让当前线程等待，支持唤醒，支持线程中断
  4. void awaitUninterruptibly();方法会释放锁，让当前线程等待，支持唤醒，不支持线程中断
  5. long awaitNanos(long nanosTimeout) throws InterruptedException;参数为纳秒，此方法会释放锁，让当前线程等待，支持唤醒，支持中断。超时以后返回的，结果为负数；超时以前返回的，结果为正数（表示返回时距离超时时间相差的纳秒数）
  6. boolean await(long time, TimeUnit unit) throws InterruptedException;方法会释放锁，让当前线程等待，支持唤醒，支持中断。超时以后返回的，结果为false；超时以前返回的，结果为true
  7. boolean awaitUntil(Date deadline) throws InterruptedException;参数表示超时的截止时间点，方法会释放锁，让当前线程等待，支持唤醒，支持中断。超时以后返回的，结果为false；超时以前返回的，结果为true
  8. void signal();会唤醒一个等待中的线程，而后被唤醒的线程会被加入同步队列，去尝试获取锁
  9. void signalAll();会唤醒全部等待中的线程，将全部等待中的线程加入同步队列，而后去尝试获取锁

java高并发系列连载中，总计估计会有四五十篇文章，能够关注公众号：javacode2018，获取最新文章。

java高并发系列 - 第13天:JUC中的Condition对象

相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程