大家好，今天我们给大家介绍一个多线程设计模式的一个概念，我们平时业务代码写得比较多，因此，如果刚上手写比较复杂多线程代码，很有可能会埋下一些坑，而这些坑一时之间都是很难发现，需要经过严格测试，甚至上线运行之后才会在生产环境显现出来。大家应该听过面向对象编程的23种设计模式吧，它就是在特定场景下提供针对某一问题的可复用解决方案，而多线程设计模式是在多线程编程领域的设计模式。今天给大家介绍其中一个设计模式：Guarded Suspension。
Guarded Suspension主要是用来解决线程协作的一些问题，其核心思想是某个线程执行特定的操作前需要满足一定条件，条件未满足则暂挂线程，处于WAITING状态，直到条件满足该线程继续执行。说到这里，大家是不是想到了wait/notify了，是的，线程的挂起和唤醒功能可以直接使用wait/notify直接实现，但除非是这方面的熟手，不然总会因为忽略了一些技术细节而犯错，而且这些重复代码散落在系统各处，往往增加了维护成本，提高了出错的概率。大家现在还能快速回忆起wait/notify的一些值得注意的编程细节吗？
比如：
最著名的是线程过早唤醒问题，当一个线程由于调用了notifyAll而醒来时，并不意味着它的保护条件是成立的，其中有各种原因，如wait方法可以“假装”返回；从线程被唤醒到wait重新获取锁的时间段内，其他线程已获取了锁并修改了保护条件中的状态；由于一个条件队列与多个保护条件相关，假设A在条件队列等待保护条件a，当B线程因为同一条件队列相关的另一个保护条件b变成真，就会调用notifyAll或者notify，唤醒了A线程，但该线程相关的保护条件a并没有成真。
因此，每次线程从wait中唤醒时，都必须再次测试保护条件是否成立，我们通常在一个循环中调用wait，相关代码的标准形式如下：
synchronized(lock){
while(!conditionPredicate){
lock.wait();
}
}
另外在实现的过程中，还有信号丢失、内存可见性、锁泄漏等各种技术细节需要我们把控，而Guarded Suspension 帮助我们把这些技术细节封装起来，统一处理，增强了代码的可复用性和可维护性。
现在来看下面这段简单的代码，描述的主要是点外卖的一个逻辑，外卖没送到之前，我们一直处于等待状态，等外卖送到，我们收到通知，就可以开吃了，我们总是避免不了去实现wait/notify一类的代码：
public class TakeOut {
private boolean foodArrived = false;
//开吃
public void eat() throws InterruptedException {
synchronized(this){
while(!foodArrived){
wait();
}
}
System.out.println("wowowo.");
}
//外卖小哥
public void foodGuy(){
synchronized(this){
System.out.println("food arrived");
this.foodArrived = true;
notifyAll();
}
}
public static void main(String[] args) throws InterruptedException {
TakeOut takeOut = new TakeOut();
Thread t = new Thread(new Runnable() {
@Override
public void run() {
try {
takeOut.eat();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
t.start();
final Timer timer = new Timer();
// 延迟50ms调用helper.stateChanged方法
timer.schedule(new TimerTask() {
@Override
public void run() {
takeOut.foodGuy();
timer.cancel();
}
}, 500, 100);
}
}
重点关注eat()和foodGuy(),在方法内部实现了wait/notify，而通常这容易犯错，有什么办法能将这些技术细节封装起来，而我们平时只要实现一些业务逻辑就可以了呢？Guarded Suspension给我们提供了一个思路，它指定了几个角色，让这些角色各司其职，而这些角色中，有些是需要开发者实现接口，有些则是可复用的代码。我们再来浏览下面这段代码，这里，开发者不需要实现关于wait/notify的技术细节，所有这些都封装在了Blocker中。
public class TakeOut2 {
private static class Helper {
private volatile boolean foodArrived = false;
private final Predicate foodArrivedNow = new Predicate() {
@Override
public boolean evaluate() {
return foodArrived;
}
};
private final Blocker blocker = new ConditionVarBlocker();
public void eat() {
//await之后的目标操作
GuardedAction<String> ga = new GuardedAction<String> (foodArrivedNow) {
@Override
public String call() throws Exception {
System.out.println("wowowo.");
return "wowowo";
}
};
try {
blocker.callWithGuard(ga);
} catch (Exception e) {
e.printStackTrace();
}
}
public void foodArrived() {
try {
blocker.signalAfter(new Callable<Boolean>() {
//状态更新操作
@Override
public Boolean call() throws Exception {
foodArrived = true;
System.out.println("food arrived");
return Boolean.TRUE;
}
});
} catch (Exception e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) throws InterruptedException {
final Helper helper = new Helper();
Thread t = new Thread(new Runnable() {
@Override
public void run() {
helper.eat();
}
});
t.start();
final Timer timer = new Timer();
// 延迟50ms调用helper.stateChanged方法
timer.schedule(new TimerTask() {
@Override
public void run() {
helper.foodArrived();
timer.cancel();
}
}, 500, 100);
}
}
这里应用开发者需要实现三个角色：
GuardedObject: 这里就是指内部类Helper，包含了受保护方法eat()和改变GuardedObject实例状态的方法foodArrived()。
ConcretePredicate：实现具体的保护条件，这里是
private final Predicate foodArrivedNow = new Predicate() {
@Override
public boolean evaluate() {
return foodArrived;
}
};
ConcreteGuardedAction:具体的目标动作及关联的保护条件
GuardedAction<String> ga = new GuardedAction<String> (foodArrivedNow) {
@Override
public String call() throws Exception {
System.out.println("wowowo.");
return "wowowo";
}
};
我们看到，有关wait/notify的代码都被封装在了Broker中，而其中的Blocker接口，可以我们自己实现，也可以使用已有实现，这里的实现是ConditionVarBlocker类，它是基于Condition类和ReentrantLock类实现的, 上面的例子用到了callWithGuard和signalAfter两方法，分别接收由应用开发者实现的GuardedAction和stateOperation，前者用于执行带保护条件的目标动作，后者用于更改状态动作的执行。
public class ConditionVarBlocker implements Blocker {
private final Lock lock;
private final Condition condition;
private final boolean allowAccess2Lock;
public ConditionVarBlocker(Lock lock) {
this(lock, true);
}
private ConditionVarBlocker(Lock lock, boolean allowAccess2Lock) {
this.lock = lock;
this.allowAccess2Lock = allowAccess2Lock;
this.condition = lock.newCondition();
}
public ConditionVarBlocker() {
this(false);
}
public ConditionVarBlocker(boolean allowAccess2Lock) {
this(new ReentrantLock(), allowAccess2Lock);
}
public Lock getLock() {
if (allowAccess2Lock) {
return this.lock;
}
throw new IllegalStateException("Access to the lock disallowed.");
}
public <V> V callWithGuard(GuardedAction<V> guardedAction) throws Exception {
lock.lockInterruptibly();
V result;
try {
final Predicate guard = guardedAction.guard;
while (!guard.evaluate()) {
Debug.info("waiting...");
condition.await();
}
result = guardedAction.call();
return result;
} finally {
lock.unlock();
}
}
public void signalAfter(Callable<Boolean> stateOperation) throws Exception {
lock.lockInterruptibly();
try {
if (stateOperation.call()) {
condition.signal();
}
} finally {
lock.unlock();
}
}
public void broadcastAfter(Callable<Boolean> stateOperation) throws Exception {
lock.lockInterruptibly();
try {
if (stateOperation.call()) {
condition.signalAll();
}
} finally {
lock.unlock();
}
}
public void signal() throws InterruptedException {
lock.lockInterruptibly();
try {
condition.signal();
} finally {
lock.unlock();
}
}
}
Broker接口定义如下：
public interface Blocker {
/**
* 在保护条件成立时执行目标动作，否则阻塞当前线程，直到保护条件成立。
* @param guardedAction 带保护条件的目标动作
* @return
* @throws Exception
*/
<V> V callWithGuard(GuardedAction<V> guardedAction) throws Exception;
/**
* 执行stateOperation所指定的操作后，决定是否唤醒本Blocker
* 所暂挂的所有线程中的一个线程。
*
* @param stateOperation
* 更改状态的操作，其call方法的返回值为true时，该方法才会唤醒被暂挂的线程
*/
void signalAfter(Callable<Boolean> stateOperation) throws Exception;
void signal() throws InterruptedException;
/**
* 执行stateOperation所指定的操作后，决定是否唤醒本Blocker
* 所暂挂的所有线程。
*
* @param stateOperation
* 更改状态的操作，其call方法的返回值为true时，该方法才会唤醒被暂挂的线程
*/
void broadcastAfter(Callable<Boolean> stateOperation) throws Exception;
}
这里注意，如果你想使用指定的Lock实例，可以在ConditionVarBlocker传入一个，而不要在外部使用，避免不必要的嵌套同步。
你可以尝试着自己实现一个Broker，这似乎是一劳永逸的事情。这里补充一个小知识点，就是Condition与wait/notify的区别。每个对象都可以作为一个锁，而每个对象也同样可以作为一个条件队列，它使得一组线程能通过某种方式等待特定的条件成真，就像一个条件对列和一个内置锁（synchronized）关联一样，每一个Condition都和一个Lock关联，它提供了比内置条件队列更丰富的功能，如条件队列可以是中断或不可中断的，基于时限的等待。
另外，一个内置锁只能有一个相关联的条件队列，多个线程可能在同一个条件队列上等待不同的保护条件，并且在最常见的加锁模式下公开条件队列对象，这使得我们notifyAll时无法满足所有等待线程为同一类型的需求，而对于Lock，可以有任意数量的Condition对象，这样就可以将保护条件分开放到多个等待线程集中，更容易满足单次通知的要求。在Condition对象中，与wait，notify，notifyAll方法对应的分别是await，signal，signalAll。
欢迎扫码关注公众号java达人：
drjava
网友评论