多线程设计模式——Guarded Suspension（保护性暂挂模式）

这些都是根据我最近看的《Java实战指南多线程编程（设计模式篇）》所得整理。

模式名称

Guarded Suspension（保护性暂挂模式）

模式面对的问题

多线程编程中，往往将一个任务分解为不同的部分，将其交由不同的线程来执行，这些线程相互协作的时候，会出现一个线程等待另一个线程一定操作过后才能进行的情景，这个时候就需要这个线程退出执行。

解决方法

有一个类GuardedObject包含受保护方法的对象，存放被保护对象的方法和被保护对象的唤醒方法。一个类GuardedAction抽象目标动作，并关联了目标动作锁需的保护条件.一个ConcreteGuardedAction应用程序锁体现的具体目标动作及其关联的保护条件，一个Predicate抽象保护条件，ConCretePredicate应用程序锁实现的具体保护条件，Blocker负责执行受保护的方法所在线程的挂起和唤醒。并执行ConcreteGuardedAction锁执行的目标操作。ConditionVarBlocker给予Java条件变量实现的Blocker.

例子代码

这也是书上的代码，需求是：某系统有个告警功能模块，改模块的主要功能是将其接收到的告警信息发送给告警服务器。

AlarmAgent类

public class ALarmAgent {
//用于记录AlarmAgent
private volatile boolean connectedToServer = false;
//模式角色：GuardedSuspension.Predicate
private final Predicate agentConnected = new Predicate(){
@Override
public boolean evaluate(){
return connectedToServer;
}
};
//模式角色GuaededSuspension.Blocker
private final Blocker blocker = new ConditionVarBlocker();
//心跳定时器
private final Timer heartbeatTimer = new Timer(true);

//发送告警信息
public void sendAlarm(final AlarmInfo alarm) throws Exception {
GuardedAction<Void> guardedAction = new GuardedAction<Void>(agentConnected){
public Void call()throws Exception{
doSendAlarm(alarm);
return null;
}
};
blocker.callWithGuard(guardedAction);
}

//通过网络链接讲告警信息发送给告警服务器
private void doSendAlarm(AlarmInfo alarm){
//省略其他与设计模式无关的代码
Debug.info("sending alarm"+alarm);
//模拟发送告警值服务器的耗时
try{
Thread.sleep(50);
}catch(Exception e){

}
}

public void init(){
//省略其他与设计模式无关的代码
//告警服务器连接线程
Thread connectingThread = new Thread(new ConnectingTask());

connectingThread.start();

heartbeatTimer.schedule(new HeartbeatTask(), 60000,2000);
}

public void disconnect(){
//省略其他与设计模式无关的代码
Debug.info("disconnected from alarm server.");
connectedToServer = false;
}

protected void onConnected(){
try{
blocker.singalAfter(new Callable<Boolean>(){
@Override
public Boolean call(){
connectedToServer = true;
Debug.info("connnected to server");
return Boolean.TRUE;
}
});
}catch(Exception e){
e.printStackTrace();
}
}

protected void onDisconnected(){
connectedToServer = false;
}

//负责与告警服务器建立网络链接
private class ConnectingTask implements Runnable{
@Override
public void run(){
//省略其他与设计模式无关的代码
//模拟连接操作耗时
try{
Thread.sleep(100);
}catch(InterruptedException e){
;
}
}
}

//心跳定时任务：定时检车与该井服务器的链接是否正常，发现链接异常后自动重新链接
private class HeartbeatTask extends TimerTask{
@Override
public void run(){
//省略其他与设计模式无关的代码
if(!testConnection()){
onDisconnected();
reconnect();
}
}
private boolean testConnection(){
//省略其他与设计模式无关的代码
return true;
}

private void reconnect(){
ConnectingTask connectingThread = new ConnectingTask();
//直接在心跳定时器线程中执行
connectingThread.run();
}
}
}

暂挂的sendAlarm

public interface Predicate {
boolean evaluate();
}

public abstract class GuardedAction<V> implements Callable<V>{
protected final Predicate guard;
public GuardedAction(Predicate guard){
this.guard = guard;
}
}

public interface Blocker {
//在保护条件城里时执行目标动作；否则阻塞当前线程，知道保护条件成立
<V> V callWithGuard(GuardedAction<V>guardedAction) throws Exception;
//执行 stateOperation所制定的操作后，决定是否唤醒本Blocker
void signalAfter(Callable<Boolean>stateOperation) throws Exception;
void signal() throws InterruptedException;
//执行stateOperation所制定的操作后，决定是否唤醒本Blocker
void broadcastAfter(Callable<Boolean> stateOperation) throws Exception;
}

public class ConditionVarBlocker implements Blocker{
private final Lock lock;

private final Condition condition;

public ConditionVarBlocker(Lock lock){
this.lock = lock;
this.condition = lock.newCondition();
}

public ConditionVarBlocker(){
this.lock = new ReentrantLock();
this.condition = lock.newCondition();
}

public <V> V callWithGuard(GuardedAction<V> guardedAction) throws Exception {
lock.lockInterruptibly();
V result;
try{
final Predicate guard = guardedAction.guard;
while(!guard.evaluate()){
condition.await();
}
result = guardedAction.call();
return result;
}finally{
lock.unlock();
}
}

public void broadcastAfter(Callable<Boolean> stateOperation)
throws Exception {
lock.lockInterruptibly();
try{
if(stateOperation.call()){
condition.signal();
}
}finally{
lock.unlock();
}
}

public void signal() throws InterruptedException {
lock.lockInterruptibly();
try{
condition.signal();
}finally{
lock.unlock();
}
}

public void signalAfter(Callable<Boolean> stateOperation) throws Exception {
lock.lockInterruptibly();
try{
if(stateOperation.call()){
condition.signalAll();
}
}finally{
lock.unlock();
}

}
}

需要关注的问题

内存可见性和锁泄露

线程被过早唤醒

嵌套监视器锁死

可能增加JVM垃圾回收的负担