Java并发-19.Condition接口
- Java对象都有一组监视器方法包括wait()、wait(long timeout)、notify()、notifyAll()方法，与synchronized关键字配合，实现等待/通知模式
- Condition接口也是提供监视器方法，和Lock配合实现等待/通知模式
- Condition接口支持等待多个队列
- Condition接口支持当前线程释放锁并进入等待状态，在等待状态中不响应中断
- Condition接口支持当前线程释放锁并进入等待状态到将来的某个时间
1.Condition接口示例
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* 使用Condition的示例
*
* @author pengjunzhe
*/
public class ConditionUseCase {
Lock lock = new ReentrantLock();
Condition condition = lock.newCondition();
public void conditionWait() throws InterruptedException {
lock.lock();
try {
condition.await();
} finally {
lock.unlock();
}
}
public void conditionSingnal() throws InterruptedException {
lock.lock();
try {
condition.signal();
} finally {
lock.unlock();
}
}
}一般而言，都会将Condition变量作为成员变量。当调用await方法后，当前线程会释放锁并进入Condition变量的等待队列，而其他线程调用signal方法后，通知正在Condition变量等待队列的线程从await方法返回，并且在返回前已经获得了锁。
Condition方法的描述：
方法 | 描述 |
|---|---|
void await() throws InterruptedException | 当前线程进入等待状态，直到被通知（signal）或者被中断时，当前线程进入运行状态，从await()返回； |
void awaitUninterruptibly() | 当前线程进入等待状态，直到被通知，对中断不做响应； |
long awaitNanos(long nanosTimeout) throws InterruptedException | 在接口1的返回条件基础上增加了超时响应，返回值表示当前剩余的时间，如果在nanosTimeout之前被唤醒，返回值 = nanosTimeout - 实际消耗的时间，返回值 <= 0表示超时； |
boolean await(long time, TimeUnit unit) throws InterruptedException | 同样是在接口1的返回条件基础上增加了超时响应，与接口3不同的是：可以自定义超时时间单位；返回值返回true/false，在time之前被唤醒，返回true，超时返回false。 |
boolean awaitUntil(Date deadline) throws InterruptedException | 当前线程进入等待状态直到将来的指定时间被通知，如果没有到指定时间被通知返回true，否则，到达指定时间，返回false； |
void signal() | 唤醒一个等待在Condition上的线程； |
void signalAll() | 唤醒等待在Condition上所有的线程。 |
一个有界队列的例子，当队列为空，队列的获取操作阻塞获取线程，直到队列中有新增元素，队列满时，队列的插入操作会阻塞线程，直到队列出现空位：
package com.junzerg.threads;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* @author pengjunzhe
*/
public class BoundedQueue<T> {
private Object[] items;
// 添加的下表，删除的下标，当前数组数量
private int addIndex, removeIndex, count;
private Lock lock = new ReentrantLock();
private Condition notEmpty = lock.newCondition();
private Condition notFull = lock.newCondition();
public BoundedQueue(int size) {
items = new Object[size];
}
/**
* 添加一个元素，数组满，则添加线程进入等待状态，直到有空位
*/
public void add(T t) throws InterruptedException {
lock.lock();
try {
while (count == items.length) {
notFull.await();
}
items[addIndex] = t;
if (++addIndex == items.length) {
addIndex = 0;
}
++count;
notEmpty.signal();
} finally {
lock.unlock();
}
}
/**
* 删除一个元素，如果数组为空，则删除线程进入等待状态，直到有心元素添加进来
*/
@SuppressWarnings("unchecked")
public T remove() throws InterruptedException {
lock.lock();
try {
while (count == 0) {
notEmpty.await();
}
Object x = items[removeIndex];
if (++removeIndex == items.length) {
removeIndex = 0;
}
--count;
notFull.signal();
return (T) x;
} finally {
lock.unlock();
}
}
}