详解Java并发之Condition

Condition是什么？

Condition是Java并发包中的一个接口，它是对传统Object.wait()和Object.notify()方法的增强，可以更灵活地实现线程的等待和通知。

创建一个Condition对象

创建Condition对象通常是在Lock对象的基础上创建的，可以通过Lock接口的newCondition()方法创建。例如下面的示例：

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

Condition的方法

Condition接口有三个常用的方法：

1. await()：等待，进入等待队列，释放当前线程获得的锁，直到被其它线程发出唤醒信号或被中断。
2. signal()：唤醒一个等待在Condition上的线程，如果有多个线程在等待，则随机选择一个。
3. signalAll()：唤醒所有等待在Condition上的线程。

Condition的基本使用

下面是一个基本的Condition使用示例，该程序启动了两个线程，线程A打印0~4，线程B打印A~E，两个线程交替执行。可以看到，线程的等待和唤醒是通过Condition对象来实现的。

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

public class ConditionDemo {
    private int state = 0;
    private Lock lock = new ReentrantLock();
    private Condition condition = lock.newCondition();

    public static void main(String[] args) throws InterruptedException {
        final ConditionDemo demo = new ConditionDemo();
        Thread threadA = new Thread(new Runnable() {
            @Override
            public void run() {
                demo.printNumbers();
            }
        });
        Thread threadB = new Thread(new Runnable() {
            @Override
            public void run() {
                demo.printLetters();
            }
        });
        threadA.start();
        threadB.start();
        threadA.join();
        threadB.join();
    }

    private void printNumbers() {
        lock.lock();
        try {
            for (int i = 0; i < 5; i++) {
                while (state % 2 == 1) {
                    condition.await();
                }
                System.out.print(i);
                state++;
                condition.signal();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    private void printLetters() {
        lock.lock();
        try {
            for (char c = 'A'; c <= 'E'; c++) {
                while (state % 2 == 0) {
                    condition.await();
                }
                System.out.print(c);
                state++;
                condition.signal();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }
}

Condition的应用实例

下面再来看一下Condition的一个应用实例，实现一个有界队列。这个队列有两个方法：put()方法往队列中放入数据，如果队列已满则等待；take()方法从队列中取出数据，如果队列为空则等待。

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

public class BoundedQueue<T> {
    private Queue<T> queue;
    private int maxSize;
    private Lock lock = new ReentrantLock();
    private Condition notEmpty = lock.newCondition();
    private Condition notFull = lock.newCondition();

    public BoundedQueue(int maxSize) {
        this.maxSize = maxSize;
        queue = new LinkedList<T>();
    }

    public void put(T element) throws InterruptedException {
        lock.lock();
        try {
            while (queue.size() == maxSize) {
                notFull.await();
            }
            queue.offer(element);
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    public T take() throws InterruptedException {
        lock.lock();
        try {
            while (queue.isEmpty()) {
                notEmpty.await();
            }
            T element = queue.poll();
            notFull.signal();
            return element;
        } finally {
            lock.unlock();
        }
    }
}

可以测试一下这个有界队列的效果：

public class BoundedQueueDemo {
    public static void main(String[] args) {
        final BoundedQueue<Integer> queue = new BoundedQueue<Integer>(5);
        Thread threadA = new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < 10; i++) {
                    System.out.println("Put: " + i);
                    try {
                        queue.put(i);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        Thread threadB = new Thread(new Runnable() {
            @Override
            public void run() {
                for (int i = 0; i < 10; i++) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    try {
                        System.out.println("Take: " + queue.take());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        threadA.start();
        threadB.start();
    }
}

这个程序会启动两个线程，线程A从0开始往队列中放入10个整数，线程B每隔1秒钟从队列中取出一个整数并打印出来。可以看到，由于队列有界，当队列满了时，线程A会等待线程B来取走数据，反之亦然。