J.U.C之Condition详解

Java 评论

一、由一道算法题开聊

我们提供一个类：

1
2
3
4
5
6
7
8
9
10
11
12
13
class FooBar {
  public void foo() {
    for (int i = 0; i < n; i++) {
      System.out.print("foo");
    }
  }

  public void bar() {
    for (int i = 0; i < n; i++) {
      System.out.print("bar");
    }
  }
}

两个不同的线程将会共用一个FooBar实例。其中一个线程将会调用foo()方法，另一个线程将会调用bar()方法。请设计修改程序，以确保 “foobar” 被输出 n 次。

示例 1:

1
2
3
输入: n = 1
输出: "foobar"
解释: 这里有两个线程被异步启动。其中一个调用 foo() 方法, 另一个调用 bar() 方法，"foobar" 将被输出一次。

示例 2:

1
2
3
输入: n = 2
输出: "foobarfoobar"
解释: "foobar" 将被输出两次。
  • 解析1——方法互调

由题意可以知道，两个线程分别调用foo()和bar()方法，按照输出要求需要使的foo()方法在bar()方法之前执行，这样我们能否在foo()方法中调用bar()方法，然后再bar()方法中调用foo()方法？我们试一下，修改代码为：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public class FooBar {
  private int     n;
  
  public FooBar(int n) {
    this.n = n;
  }
  
  public void foo() {
    for (int i = 0; i < n; i++) {
      System.out.print("foo");
      bar();
    }
  }

  public void bar() {
    for (int i = 0; i < n; i++) {
      System.out.print("bar");
      foo();
    }
  }
}

测试方法：

1
2
3
4
5
public static void main(String[] args) {
   FooBar fooBar = new FooBar(2);
   new Thread(() -> fooBar.foo()).start();
   new Thread(() -> fooBar.bar()).start();
}

执行结果：

image-20200508010813617

发生了栈溢出，这是因为我们在两个方法之间互相调用，不断的将方法压入栈中，最终造成栈溢出，整个过程如下：

image-20200508012241430

可见此办法不可行。

  • 解析2——循环阻塞

使用一个临时变量来控制方法的执行进度，然后通过死循环来临时中断方法的执行，修改代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
class FooBar {
    private int     n;

    private boolean flag = true;

    public FooBar(int n) {
        this.n = n;
    }

    public void foo() {
        for (int i = 0; i < n; i++) {
            while (!flag) {}
            System.out.print("foo");
            flag = false;
        }
    }

    public void bar() {
        for (int i = 0; i < n; i++) {
            while (flag) {}
            System.out.print("bar");
            flag = true;
        }
    }
}

执行结果：

1
foobarfoobar

好像执行成功了，但是当我们把n放大时，会出现两个线程都进入到while循环之中，之所以出现这种情况是因为变量flag在线程之间出现了数据不同步的问题。针对这个问题，Java提供了关键字volatile来保证变量在线程之间的修改同步，这里就不介绍volatile了，只要我们把变量flag修改为private volatile boolean flag = true;即可解决。

  • 解析3——Lock+Condition+临时变量

以上方案我们都没有使用到锁，因为题目已经定义好了方法，所以我们给方法加synchronized方案不可取，并且我们在输出的时候要保证foo()方法先执行，仅通过synchronized来保证顺序，实现上有些困难，所以我们采用J.U.C包下的Lock类，通过Condition类来控制方法的执行和等待，然后通过一个临时变量flag控制方法的执行顺序，修改代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class FooBar {
    private int       n;

    private Lock      lock      = new ReentrantLock();
    private Condition condition = lock.newCondition();
    private boolean   flag      = false;

    public FooBar(int n) {
        this.n = n;
    }

    public void foo(Runnable printFoo) throws InterruptedException {

        lock.lock();
        for (int i = 0; i < n; i++) {
            while (flag) {
                condition.await();
            }
            printFoo.run();
            flag = true;
            condition.signal();
        }
        lock.unlock();
    }

    public void bar(Runnable printBar) throws InterruptedException {

        lock.lock();
        for (int i = 0; i < n; i++) {
            while (!flag) {
                condition.await();
            }
            printBar.run();
            flag = false;
            condition.signal();
        }
        lock.unlock();
    }
}

执行结果：

1
foobarfoobar

此方案中的参数flag并没有被volatile关键字修饰，且未出现方案2中的修改不同步的问题，这是因为我们在修改flag的时候被加锁了，所以同时只能被一个线程修改。在for循环中通过while循环来控制线程的状态，这里其实只要判断flag的值来决定是否需要将线程挂起，那么我们用if来判断是否可以？

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
class FooBar {
    private int       n;

    private Lock      lock      = new ReentrantLock();
    private Condition condition = lock.newCondition();
    private boolean   flag      = false;

    public FooBar(int n) {
        this.n = n;
    }

    public void foo() throws InterruptedException {

        lock.lock();
        for (int i = 0; i < n; i++) {
            if (flag) {
                condition.await();
            }
            System.out.print("foo");
            flag = true;
            condition.signal();
        }
        lock.unlock();
    }

    public void bar() throws InterruptedException {

        lock.lock();
        for (int i = 0; i < n; i++) {
            if (!flag) {
                condition.await();
            }
            System.out.print("bar");
            flag = false;
            condition.signal();
        }
        lock.unlock();
    }
}

执行结果正常，所以这里我们不论是用while还是用if都可以，只要控制好flag的值即可。

二、Condition概述

在上述方案3中，我们使用到了Lock和Condition两个类，Lock我们都知道，它是控制多线程访问共享资源的一个工具，此类在jdk1.5之后提供，它在使用上有几个原则：

  • 要访问共享资源，必须先获得锁
  • 同一时刻只能有一个线程获取到锁
  • 一般情况下Lock是排它锁
  • 但是也可以使用ReadWriteLock对共享资源进行并发访问

Lock的使用方式：

1
2
3
4
5
6
7
8
9
10
void m() {
  Lock lock = new ReentrantLock();
  lock.lock();
  try {
      //TODO 访问共享资源
  } finally {
      //释放锁
      l.unlock();
  }
}

Lock的源码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
public interface Lock {

    /**
     * 获取锁.调用该方法的线程会得到锁
     */
    void lock();

    /**
     * 可中断的获取锁，该方法可以响应中断(可以中断当前线程)
     */
    void lockInterruptibly() throws InterruptedException;

    /**
     * 尝试非阻塞地获取锁
     * 调用后方法立即返回:
     *	如果能正常获取到锁，则立即返回true
     *	如果没能正常获取到锁，则立即返回false
     */
    boolean tryLock();

    /**
     * 在指定的超时时间内获取锁
     * 如果锁可以立即获得，则直接返回true，否则，在以下三种情况下会返回:
     * 		1. 在指定的超时时间内获取到了锁(true)
     * 		2. 在指定的超时时间内线程被interrupt()了(false)
     * 		3. 超时(false)
     */
    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;

    /**
     * 释放锁
     */
    void unlock();

    /**
     * 返回和当前Lock对象绑定的Condition实例
     */
    Condition newCondition();
}

在jdk1.5之前，我们如果想要实现线程的等待和恢复只能通过Object的相关wait和notify方法与synchronized关键字配合使用，Object类中提供的方法如下：

1
2
3
4
5
java.lang.Object.wait()
java.lang.Object.wait(long)
java.lang.Object.wait(long, int)
java.lang.Object.notify()
java.lang.Object.notifyAll()

而从jdk1.5开始，我们通过java.util.concurrent.locks.Condition接口来替代Object来实现类似功能。在整个juc中的锁世界中，AQS是所有锁的基础，主要的实现在AQS中的ConditionObject类，ConditionObject的等待队列是一个FIFO队列，队列的每个节点都是等待在Condition对象上的线程的引用，在调用Condition的await()方法之后，线程释放锁，构造成相应的节点进入等待队列等待

Condition源码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
public interface Condition {

    /**
     * 使当前线程进入等待状态并释放锁，直到接收到唤醒信号或线程被中断
     */
    void await() throws InterruptedException;

    /**
     * 使当前线程进入等待状态并释放锁，直到接收到唤醒信号，
     * 如果在等待过程中线程被中断，此方法会继续执行，线程扔处于等待状态
     */
    void awaitUninterruptibly();

    /**
     * 和wait()方法相同，但是在等待时间（纳秒）内未收到通知，则终止等待并返回0或负数
     */
    long awaitNanos(long nanosTimeout) throws InterruptedException;

    /**
     * 和wait()方法相同，但是在等待时间（指定时间和单位）内未收到通知，则终止等待并返回0或负数
     */
    boolean await(long time, TimeUnit unit) throws InterruptedException;

    /**
     * 和wait()方法相同，但是在指定某个时间点未收到通知，则终止等待并返回0或负数
     */
    boolean awaitUntil(Date deadline) throws InterruptedException;

    /**
     * 唤醒一个在await()等待队列中的线程
     */
    void signal();

    /**
     * 唤醒所有在await()等待队列中的线程
     */
    void signalAll();
}

串一串浪人

断舍离