标签：operation this false 查询 upd lock rup return rtb
保护性暂挂模式，也称为Guarded Suspension模式，指的是当前线程在执行某个任务之前，需要检查某一条件，只有在该条件成立的情况下，当前线程才可以继续往下执行当前任务。顾名思义，保护性暂挂模式是一种广义的概念，其主要载体有两个：预备条件和任务，在任何需要使用预先检查的情况中都可以使用保护性暂挂模式。
如下是类图中各个角色定位的描述：
比如我们会遇到这种场景，在进行某些操作时，比如通过elasticsearch服务器进行查询或更新操作，我们需要连接es服务器，而在es服务器连接上之前，所有的查询和更新操作都是需要被阻塞的。即使在服务器连接上之后，我们也需要经常对服务器进行心跳测试，以检查与服务器的连接是否还存活在，如果不存活，则还是需要继续阻塞其余的操作，并且尝试重新连接es服务器，这种情况我们就可以使用到保护性暂挂模式。保护性条件即是与es服务器的连接还存活在，如果不存活则需要挂起所有尝试连接服务器执行任务的线程，并且当前线程会尝试连接服务器。如下是示例代码：
public class ElasticSearchAgent {
private volatile boolean connectedToServer = false;
private final Predicate agentConnected = () -> connectedToServer;
private final Blocker blocker = new ConditionVarBlocker();
private final Timer heartbeatTimer = new Timer(true);
public void update(final UpdateCondition condition) throws Exception {
GuardedAction<Void> guardedAction = new GuardedAction<Void>(agentConnected) {
@Override
public Void call() {
doUpdate(condition);
return null;
}
};
blocker.callWithGuard(guardedAction);
}
private void doUpdate(UpdateCondition condition) {
try {
TimeUnit.MICROSECONDS.sleep(20); // 模拟进行更新
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public void init() {
Thread connectingThread = new Thread(new ConnectingTask());
connectingThread.start();
heartbeatTimer.schedule(new HeartBeatTask(), 60000, 2000);
}
public void disconnect() {
connectedToServer = false;
}
protected void onConnected() {
try {
blocker.signalAfter(() -> {
connectedToServer = true;
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) {}
onConnected();
}
}
private class HeartBeatTask extends TimerTask {
@Override
public void run() {
if (!testConnection()) {
onDisconnected();
reconnect();
}
}
private boolean testConnection() {
return true;
}
private void reconnect() {
ConnectingTask connectingTask = new ConnectingTask();
connectingTask.run();
}
}
}可以看到，在进行update()操作时，首先会创建一个GuardedAction对象，真正的更新操作是在该对象中进行的，这里的保护性条件是通过一个volatile类型的变量connectedToServer来控制的，如果当前与es服务器的连接还存活在，则该变量置为true。HeartBeatTask是一个定时任务，在60s延迟之后每隔2s会向服务器发送心跳测试，以检查连接是否存活，如果不存活，则会将connectedToServer变量置为false，并且会尝试连接服务器。在init()方法中首先会创建一个连接服务器的任务，以保证服务器连接在初始时的可用状态，并且其还会启动心跳测试的定时任务。如下是Blocker和ConditionVarBlocker的实现代码：
public interface Blocker {
<V> V callWithGuard(GuardedAction<V> guardedAction) throws Exception;
void signalAfter(Callable<Boolean> stateOperation) throws Exception;
void signal() throws InterruptedException;
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();
}
@Override
public <V> V callWithGuard(GuardedAction<V> guardedAction) throws Exception {
lock.lockInterruptibly();
try {
final Predicate guard = guardedAction.guard;
while (!guard.evaluate()) {
condition.await();
}
return guardedAction.call();
} finally {
lock.unlock();
}
}
@Override
public void signalAfter(Callable<Boolean> stateOperation) throws Exception {
lock.lockInterruptibly();
try {
if (stateOperation.call()) {
condition.signal();
}
} finally {
lock.unlock();
}
}
@Override
public void signal() throws InterruptedException {
lock.lockInterruptibly();
try {
condition.signal();
} finally {
lock.unlock();
}
}
@Override
public void broadcastAfter(Callable<Boolean> stateOperation) throws Exception {
lock.lockInterruptibly();
try {
if (stateOperation.call()) {
condition.signalAll();
}
} finally {
lock.unlock();
}
}
}可以看到，ConditionVarBlocker中基本上都是模板代码，其声明了一个Lock对象和一个Condition对象，Lock对象用于对当前的先验条件检查过程进行同步处理，Condition对象则用于在先验条件不满足的情况下阻塞当前线程的。
在callWithGuard()方法中，首先会在一个循环中检查当前的先验条件是否满足，如果不满足，则使当前线程进入等待状态，如果满足，则当前线程继续执行其任务。这里需要注意的是，我们使用了while()循环用于判断先验条件是否满足，因为有可能当前线程被意外的唤醒，或者说被唤醒之后先验条件还是不满足，因而这里使用循环判断，以使当前线程在先验条件不满足的情况下继续等待。
在signalAfter()方法中，其首先调用stateOperation.call()方法，判断当前的先验条件是否满足，只有在先验条件满足的情况下才会唤醒一个等待的线程。这里stateOperation是ElasticSearchAgent传入的用于判断当前是否处于连接状态的一个条件载体。
如下是GuardedAction的实现代码：
public abstract class GuardedAction<V> implements Callable<V> {
protected final Predicate guard;
public GuardedAction(Predicate guard) {
this.guard = guard;
}
}这里GuardedAction是一个抽象类，其主要封装了一个Predicate属性。GuardedAction的主要实现在ElasticSearchAgent.guardedMethod()方法中生成的，因为具体需要执行的任务需要调用方生成，这里只是提供了一个模板方法。如下是Predicate的代码：
@FunctionalInterface
public interface Predicate {
boolean evaluate();
}这里Predicate也只是一个声明而已，其具体的实现也是在ElasticSearchAgent中，本例中主要是判断connectedToServer是否为true，即处于连接服务器的状态。
public ConditionVarBlocker(Lock lock) {
this.lock = lock;
this.condition = lock.newCondition();
}该方法用于防止ElasticSearchAgent由于某种原因而需要加锁时可能会造成嵌套监视器锁死的问题的。所谓的嵌套监视器锁死的问题指的是，如果某个线程执行依次获取了两个锁，而由于先验条件不满足，从而导致当前线程释放了内层锁从而进入等待状态，而另外的线程为了检查当前的先验条件需要获取到外层锁，这就导致了锁循环等待的问题，在等待先验条件满足的线程持有外层锁，其无法释放，而尝试改变先验条件的线程正在尝试获取外层锁，但其一直无法获取到，从而造成了死锁。这种情况下就提供了该构造方法，如果ElasticSearchAgent需要对其方法进行加锁，那么其需要通过该构造方法将锁传递给ConditionVarBlocker，这样当前线程在释放锁的时候就会将外层锁和内层锁同时释放了（因为都是同一个锁）。
标签：operation this false 查询 upd lock rup return rtb
原文地址：https://www.cnblogs.com/zhangxufeng/p/9162172.html
