码迷,mamicode.com
首页 > 编程语言 > 详细

多线程编程之保护性暂挂模式

时间:2018-06-10 12:05:47      阅读:197      评论:0      收藏:0      [点我收藏+]

标签:operation   this   false   查询   upd   lock   rup   return   rtb   

       保护性暂挂模式,也称为Guarded Suspension模式,指的是当前线程在执行某个任务之前,需要检查某一条件,只有在该条件成立的情况下,当前线程才可以继续往下执行当前任务。顾名思义,保护性暂挂模式是一种广义的概念,其主要载体有两个:预备条件和任务,在任何需要使用预先检查的情况中都可以使用保护性暂挂模式。

1. 角色描述

技术分享图片

       如下是类图中各个角色定位的描述:

  • GuardedObject:受保护的对象,供给客户端调用,用于生成并执行受保护的行为的和检查并控制状态改变的,如下是其各个方法的作用:
  • guardedMethod():生成并且执行受保护的行为;
  • stateOperation():用于检查当前状态是否满足特定的状态,从而控制受保护行为的状态;
  • Blocker:该类主要提供了一些模板方法,主要是用于调用受保护行为,或者唤醒当前由于先验条件不通过而导致等待的线程的,如下是其各个方法的作用:
  • callWithGuard(GuardedAction):该方法首先会检查GuardedAction提供的先验条件是否满足,如果不满足,则会阻塞当前线程,否则会执行GuardedAction中的任务;
  • signalAfter(Callable):该方法会先检查先验条件是否满足,如果满足先验条件则会唤醒一个正在等待的线程;
  • broadcastAfter(Callable):该方法会先检查先验条件是否满足,如果先验条件满足则会唤醒所有正在等待的线程;
  • signal:直接唤醒一个正在等待先验条件满足的线程;
  • broadcast():直接唤醒所有正在等待先验条件满足的线程;
  • GuardedAction:受保护方法的载体,并且提供了进行先验检查的条件,其各方法和属性作用如下:
  • predicate:提供了进行先验检查的条件;
  • call():提供了需要执行的任务;
  • Predicate:承载了进行先验条件检查的条件。

2. 实例演示

       比如我们会遇到这种场景,在进行某些操作时,比如通过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,即处于连接服务器的状态。

3. Guarded Suspension实现考量

  • 可以看到Guarded Suspension的实现中,Blocker、GuardedAction和Predicate只是提供的一个模板,其内主要是一些通用的代码,而和具体业务相关的代码主要在ElasticSearchAgent中,其会创建所需执行的GuardedAction对象,并且控制其所需的先验条件。Guarded Suspension将关注点进行了分离,我们在使用该模式的时候主要需要实现的也即是类似于ElasticSearchAgent的一个客户端类;
  • 在执行使用Guarded Suspension模式的时候,需要注意的是,每次执行GuardedObject.guardedMethod()方法时都会创建一个GuardedAction对象,这可能会对JVM垃圾回收造成一定的负担,因而在使用该模式时如果内存较小需要特别注意该问题;
  • 在Guarded Suspension模式中,ConditionVarBlocker的callWithGuard()和signal*()方法的执行都进行了加锁处理,这是因为ConditionVarBlocker是所有线程所共有的一个对象,其lock和condition变量是需要所有线程都一致可见的,因而这里需要对其进行加锁处理;
  • 在ConditionVarBlocker.callWithGuard()方法中,对先验条件的检查是使用一个while循环进行的,这是为了防止等待的线程被意外的唤醒,而先验条件此时还不满足,使用while循环就可以保证当前线程再次进入到等待状态;
  • 上述ConditionVarBlocker还提供了一个如下的构造方法:
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

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!