Activiti架构分析

1.2.2 PersistentObject

public interface PersistentObject {
  String getId();
  void setId(String id);
  Object getPersistentState();
}

1.2.3 DbSqlSession

//该属性存储着所有使用insert方法放入的对象
protected Mapextends PersistentObject>, List> insertedObjects = 
　　　　　　　　　　　　　　　　　　　　　　　　　　new HashMapextends PersistentObject>, List>();
//该Map结构内存储所有通过该DbSqlSession查询出来的结果，以及update方法放入的对象
protected Map, Map> cachedObjects = new HashMap, Map>();
//该属性内存储着所有将要执行的删除操作
protected List deleteOperations = new ArrayList();

public static class CachedObject {
    protected PersistentObject persistentObject;
    protected Object persistentObjectState;
}
public CachedObject(PersistentObject persistentObject, boolean storeState) {
      this.persistentObject = persistentObject;
      if (storeState) {
        this.persistentObjectState = persistentObject.getPersistentState();
      }
}

• 所有通过该DbSqlSession对象执行查询的结果对象都会生成一个对应的CachedObject对象，并且storeState参数为true
• 执行该DbSqlSession对象的update类方法，会将参数用CachedObject对象包装起来，storeState参数为false。

1. 将insertObjects列表中的元素插入到数据库
2. 将deleteOperations列表中的元素遍历执行
3. 执行方法getUpdatedObjects获得要更新的实体对象

1. 实体的getPersistentState方法不为空
2. 实体的getPersistentState方法返回对象与CachedObject存储的persistentObjectState执行equal判断，结果为false

二、Activiti设计解析-PVM执行树

2.1 核心理念

2.1.1 PVM对流程定义期的描述

• 流程节点是某一种动作表达的抽象描述。节点本身是可以嵌套的，也就是节点可以拥有子节点。
• 连线表达是不同节点之间的转移关系。一个连线只能有一个源头节点和一个目标节点。而节点本身可以有任意多的进入连线和外出连线。

2.1.2 PVM对流程运行期的描述

1. 流程节点的具体执行动作。
2. 流程执行当前处于哪一个流程节点。
3. 流程执行是如何从一个节点运行至下一个节点。
4. 流程执行如何执行流程节点定义的执行动作

public interface InterpretableExecution extends ActivityExecution, ExecutionListenerExecution, PvmProcessInstance {
  void take(PvmTransition transition);
  void take(PvmTransition transition, boolean fireActivityCompletedEvent);
｝

2.1.3 PVM综述

2.2 ActivitiImpl与作用域

1. 该ActivityImpl是可变范围，则它是一个作用域。可变范围可以理解为该节点的内容定义是可变的。比如流程定义、子流程，其内部内容是可变的。根据BPMN定义，可变范围有：流程定义，子流程，多实例，调用活动。
2. 该ActivityImpl定义了一个上下文用于接收事件。比如：具备边界事件的ActivityImpl，具备事件子流程的ActivityImpl，事件驱动网关，中间事件捕获ActivityImpl。

2.3 ExecutionEntity

• isScope：该属性为真时，意味该执行实例在执行一个具备作用域的ActivityImpl节点或者执行一个流程定义。更简单一些，意味着该实例正在执行一个作用域活动。
• isConcurrent：该属性为真时，意味与该执行实例正在执行的活动节点同属相同作用域的节点可能正并发被其他执行实例执行（比如并行网关后面的2个并行任务）。
• isActive：该属性为真时，意味该执行实例正在执行一个简单ActivityImpl（不包含其他ActivityImpl的ActivityImpl）
• isEventScope：该属性为真时，意味该执行实例是为了后期补偿而进行的变量保存所创建的执行实例。由于流程执行中的变量都需要与ExecutionEntity挂钩，而补偿是需要原始变量的快照。为了满足这个需求，创建出一个专用于此的ExecutionEntity。
• activityId：该ExecutionEntity正在执行的ActivityImpl的id。正在执行意味着几种情况：进入该节点，执行该节点动作，离开该节点。如果是等待子流程的完成，则该属性为null。

1. 单独的非作用域节点
2. 单独的作用域节点
3. 并发的非作用域节点
4. 并发的作用域节点

2.3.1 单独的非作用域节点

2.3.2 单独的作用域节点

2.3.3 并发的非作用域节点

2.3.4 并发的作用域节点

三、代码解析-流程启动

3.1 流程说明

1. 流程定义实体创建流程实例
2. 流程实例执行启动

3.1.1 流程定义实体创建流程实例

3.1.2 流程实例启动

3.2 额外补充

3.2.1 ActivityImpl的parent属性

四、代码解析-原子操作

4.1 说明

4.2 AbstractEventAtomicOperation

public abstract class AbstractEventAtomicOperation implements AtomicOperation {
 
  public boolean isAsync(InterpretableExecution execution) {
    return false;
  }
 
  public void execute(InterpretableExecution execution) {
      //获取当前执行对象的作用域对象。具体由子类提供。
    ScopeImpl scope = getScope(execution);
      //从作用域对象中获取指定事件的监听器。事件名称由子类提供。
    List exectionListeners = scope.getExecutionListeners(getEventName());
    int executionListenerIndex = execution.getExecutionListenerIndex();
 
    if (exectionListeners.size()>executionListenerIndex) {
      execution.setEventName(getEventName());
      execution.setEventSource(scope);
      ExecutionListener listener = exectionListeners.get(executionListenerIndex);
      try {
        listener.notify(execution);
      } catch (RuntimeException e) {
        throw e;
      } catch (Exception e) {
        throw new PvmException("couldn't execute event listener : "+e.getMessage(), e);
      }
      execution.setExecutionListenerIndex(executionListenerIndex+1);
      execution.performOperation(this);
 
    } else {
      execution.setExecutionListenerIndex(0);
      execution.setEventName(null);
      execution.setEventSource(null);
 
      eventNotificationsCompleted(execution);
    }
  }
 
  protected abstract ScopeImpl getScope(InterpretableExecution execution);
  protected abstract String getEventName();
  protected abstract void eventNotificationsCompleted(InterpretableExecution execution);
}

4.3 AtomicOperationProcessStart

public class AtomicOperationProcessStart extends AbstractEventAtomicOperation {
 
  @Override
  protected ScopeImpl getScope(InterpretableExecution execution) {
    return execution.getProcessDefinition();
  }
 
  @Override
  protected String getEventName() {
    return org.activiti.engine.impl.pvm.PvmEvent.EVENTNAME_START;
  }
 
  @Override
  protected void eventNotificationsCompleted(InterpretableExecution execution) {
      if (Context.getProcessEngineConfiguration() != null && Context.getProcessEngineConfiguration().getEventDispatcher().isEnabled()) {
        Map variablesMap = null;
        try {
          variablesMap = execution.getVariables();
        } catch (Throwable t) {
          // In some rare cases getting the execution variables can fail (JPA entity load failure for example)
          // We ignore the exception here, because it's only meant to include variables in the initialized event.
        }
        Context.getProcessEngineConfiguration().getEventDispatcher().dispatchEvent(
                ActivitiEventBuilder.createEntityWithVariablesEvent(ActivitiEventType.ENTITY_INITIALIZED,
                    execution, variablesMap, false));
      Context.getProcessEngineConfiguration().getEventDispatcher()
              .dispatchEvent(ActivitiEventBuilder.createProcessStartedEvent(execution, variablesMap, false));
    }
 
    ProcessDefinitionImpl processDefinition = execution.getProcessDefinition();
    StartingExecution startingExecution = execution.getStartingExecution();
    List initialActivityStack = processDefinition.getInitialActivityStack(startingExecution.getInitial());
    execution.setActivity(initialActivityStack.get(0));
    execution.performOperation(PROCESS_START_INITIAL);
  }
}

4.4 AtomicOperationProcessStartInitial

public class AtomicOperationProcessStartInitial extends AbstractEventAtomicOperation {
 
  @Override
  protected ScopeImpl getScope(InterpretableExecution execution) {
    return (ScopeImpl) execution.getActivity();
  }
 
  @Override
  protected String getEventName() {
    return org.activiti.engine.impl.pvm.PvmEvent.EVENTNAME_START;
  }
 
  @Override
  protected void eventNotificationsCompleted(InterpretableExecution execution) {
    ActivityImpl activity = (ActivityImpl) execution.getActivity();
    ProcessDefinitionImpl processDefinition = execution.getProcessDefinition();
    StartingExecution startingExecution = execution.getStartingExecution();
     //从开始节点开始的，该判断均为真。
    if (activity==startingExecution.getInitial()) {
      execution.disposeStartingExecution();
      execution.performOperation(ACTIVITY_EXECUTE);
    } else {
      List initialActivityStack = processDefinition.getInitialActivityStack(startingExecution.getInitial());
      int index = initialActivityStack.indexOf(activity);
      activity = initialActivityStack.get(index+1);
 
      InterpretableExecution executionToUse = null;
      if (activity.isScope()) {
        executionToUse = (InterpretableExecution) execution.getExecutions().get(0);
      } else {
        executionToUse = execution;
      }
      executionToUse.setActivity(activity);
      executionToUse.performOperation(PROCESS_START_INITIAL);
    }
  }
}

4.5 AtomicOperationTransitionNotifyListenerEnd

4.6 AtomicOperationTransitionNotifyListenerStart

4.3 AtomicOperationActivityExecute

public class AtomicOperationActivityExecute implements AtomicOperation {
 
  private static Logger log = LoggerFactory.getLogger(AtomicOperationActivityExecute.class);
 
  public boolean isAsync(InterpretableExecution execution) {
    return false;
  }
 
  public void execute(InterpretableExecution execution) {
    ActivityImpl activity = (ActivityImpl) execution.getActivity();
 
    ActivityBehavior activityBehavior = activity.getActivityBehavior();
    if (activityBehavior==null) {
      throw new PvmException("no behavior specified in "+activity);
    }
 
    log.debug("{} executes {}: {}", execution, activity, activityBehavior.getClass().getName());
 
    try {
        if(Context.getProcessEngineConfiguration() != null && Context.getProcessEngineConfiguration().getEventDispatcher().isEnabled()) {
          Context.getProcessEngineConfiguration().getEventDispatcher().dispatchEvent(
                  ActivitiEventBuilder.createActivityEvent(ActivitiEventType.ACTIVITY_STARTED,
                          execution.getActivity().getId(),
                          (String) execution.getActivity().getProperty("name"),
                          execution.getId(),
                          execution.getProcessInstanceId(),
                          execution.getProcessDefinitionId(),
                          (String) activity.getProperties().get("type"),
                          activity.getActivityBehavior().getClass().getCanonicalName()));
      }
 
      activityBehavior.execute(execution);
    } catch (RuntimeException e) {
      throw e;
    } catch (Exception e) {
      LogMDC.putMDCExecution(execution);
      throw new PvmException("couldn't execute activity <"+activity.getProperty("type")+" id=\""+activity.getId()+"\" ...>: "+e.getMessage(), e);
    }
  }
}

4.4 AtomicOperationTransitionDestroyScope

public class AtomicOperationTransitionDestroyScope implements AtomicOperation {
 
  private static Logger log = LoggerFactory.getLogger(AtomicOperationTransitionDestroyScope.class);
 
  public boolean isAsync(InterpretableExecution execution) {
    return false;
  }
 
  @SuppressWarnings("unchecked")
  public void execute(InterpretableExecution execution) {
    InterpretableExecution propagatingExecution = null;
 
    ActivityImpl activity = (ActivityImpl) execution.getActivity();
    /**
    * 如果当前的活动节点具备作用域。这就意味着最初的时候，有一个处于激活状态的执行实例在执行该节点，以下称初始执行实例。
    * 从这样的节点退出要考虑几种情况：
    * 一、单独的作用域节点。此时的执行树情况是：非激活的非作用域执行实例-激活的作用域执行实例（初始执行实例）。
    *    此时要离开作用域节点，首先是销毁激活的作用域执行实例（初始执行实例），激活初始执行实例的父实例，使用该父实例执行后续的出线动作。
    * 二、并行的作用域节点。此时的执行树情况是：非激活的非作用域并发执行实例-非激活的
    */
    if (activity.isScope()) {
 
      InterpretableExecution parentScopeInstance = null;
      // if this is a concurrent execution crossing a scope boundary
      if (execution.isConcurrent() && !execution.isScope()) {
        // first remove the execution from the current root
        InterpretableExecution concurrentRoot = (InterpretableExecution) execution.getParent();
        parentScopeInstance = (InterpretableExecution) execution.getParent().getParent();
 
        log.debug("moving concurrent {} one scope up under {}", execution, parentScopeInstance);
        List parentScopeInstanceExecutions = (List) parentScopeInstance.getExecutions();
        List concurrentRootExecutions = (List) concurrentRoot.getExecutions();
        // if the parent scope had only one single scope child
        if (parentScopeInstanceExecutions.size()==1) {
          // it now becomes a concurrent execution
          parentScopeInstanceExecutions.get(0).setConcurrent(true);
        }
 
        concurrentRootExecutions.remove(execution);
        parentScopeInstanceExecutions.add(execution);
        execution.setParent(parentScopeInstance);
        execution.setActivity(activity);
        propagatingExecution = execution;
 
        // if there is only a single concurrent execution left
        // in the concurrent root, auto-prune it.  meaning, the
        // last concurrent child execution data should be cloned into
        // the concurrent root. 
        if (concurrentRootExecutions.size()==1) {
          InterpretableExecution lastConcurrent = concurrentRootExecutions.get(0);
          if (lastConcurrent.isScope()) {
            lastConcurrent.setConcurrent(false);
 
          } else {
            log.debug("merging last concurrent {} into concurrent root {}", lastConcurrent, concurrentRoot);
 
            // We can't just merge the data of the lastConcurrent into the concurrentRoot.
            // This is because the concurrent root might be in a takeAll-loop.  So the
            // concurrent execution is the one that will be receiving the take
            concurrentRoot.setActivity((ActivityImpl) lastConcurrent.getActivity());
            concurrentRoot.setActive(lastConcurrent.isActive());
            lastConcurrent.setReplacedBy(concurrentRoot);
            lastConcurrent.remove();
          }
        }
 
      } else if (execution.isConcurrent() && execution.isScope()) {
        /**
        * 根据算法，这种情况不会出现。源代码中，这部分也属于todo的内容。
        */
      }
        else {
        /**
        * 这个条件是执行实例的scope属性为真。此时销毁当前的执行实例，使用其父执行实例继续后面的流程
        */
        propagatingExecution = (InterpretableExecution) execution.getParent();
        propagatingExecution.setActivity((ActivityImpl) execution.getActivity());
        propagatingExecution.setTransition(execution.getTransition());
        propagatingExecution.setActive(true);
        log.debug("destroy scope: scoped {} continues as parent scope {}", execution, propagatingExecution);
        execution.destroy();
        //删除与该执行实例相关的一切，包括：定时工作，各种任务，事件订阅，用户流程关系，最后删除自身。
        execution.remove();
      }
    } else {
      //如果离开的是一个非作用域节点，则仍然使用当前的执行实例作为下一个节点的执行实例
      propagatingExecution = execution;
    }
 
    // if there is another scope element that is ended
    ScopeImpl nextOuterScopeElement = activity.getParent();
    TransitionImpl transition = propagatingExecution.getTransition();
    ActivityImpl destination = transition.getDestination();
    /**
    * 考虑当前的节点可能是子流程或者活动调用中的节点。那么就需要离开当前的作用域范围，回到更上层的作用域下。
    * 因此需要判断目的地是否和源头节点处于同一个作用域。如果不是同一个作用域，则不断向上回溯
    */
    if (transitionLeavesNextOuterScope(nextOuterScopeElement, destination)) {
      propagatingExecution.setActivity((ActivityImpl) nextOuterScopeElement);
      propagatingExecution.performOperation(TRANSITION_NOTIFY_LISTENER_END);
    } else {
      propagatingExecution.performOperation(TRANSITION_NOTIFY_LISTENER_TAKE);
    }
  }
 
  public boolean transitionLeavesNextOuterScope(ScopeImpl nextScopeElement, ActivityImpl destination) {
    return !nextScopeElement.contains(destination);
  }
}

4.5 AtomicOperationTransitionNotifyListenerTake

public class AtomicOperationTransitionNotifyListenerTake implements AtomicOperation {
 
  private static Logger log = LoggerFactory.getLogger(AtomicOperationTransitionNotifyListenerTake.class);
 
  public boolean isAsync(InterpretableExecution execution) {
    return false;
  }
 
  public void execute(InterpretableExecution execution) {
    TransitionImpl transition = execution.getTransition();
 
    List executionListeners = transition.getExecutionListeners();
    int executionListenerIndex = execution.getExecutionListenerIndex();
    /**
    * 整个if的功能就是不断判断监听器是否被执行完毕。都执行完毕后走入到else的部分。
    */
    if (executionListeners.size()>executionListenerIndex) {
      execution.setEventName(org.activiti.engine.impl.pvm.PvmEvent.EVENTNAME_TAKE);
      execution.setEventSource(transition);
      ExecutionListener listener = executionListeners.get(executionListenerIndex);
      try {
        listener.notify(execution);
      } catch (RuntimeException e) {
        throw e;
      } catch (Exception e) {
        throw new PvmException("couldn't execute event listener : "+e.getMessage(), e);
      }
      execution.setExecutionListenerIndex(executionListenerIndex+1);
      execution.performOperation(this);
 
    } else {
        if (log.isDebugEnabled()) {
            log.debug("{} takes transition {}", execution, transition);
        }
      execution.setExecutionListenerIndex(0);
      execution.setEventName(null);
      execution.setEventSource(null);
 
      ActivityImpl activity = (ActivityImpl) execution.getActivity();
      ActivityImpl nextScope = findNextScope(activity.getParent(), transition.getDestination());
      execution.setActivity(nextScope);
 
      // Firing event that transition is being taken       
      if(Context.getProcessEngineConfiguration() != null && Context.getProcessEngineConfiguration().getEventDispatcher().isEnabled()) {
          Context.getProcessEngineConfiguration().getEventDispatcher().dispatchEvent(
                ActivitiEventBuilder.createSequenceFlowTakenEvent(ActivitiEventType.SEQUENCEFLOW_TAKEN, transition.getId(),
                        activity.getId(), (String) activity.getProperties().get("name") ,
                        (String) activity.getProperties().get("type"), 
                        activity.getActivityBehavior().getClass().getCanonicalName(),
                        nextScope.getId(), (String) nextScope.getProperties().get("name"), 
                        (String) nextScope.getProperties().get("type"), nextScope.getActivityBehavior().getClass().getCanonicalName()));
      }
 
      execution.performOperation(TRANSITION_CREATE_SCOPE);
    }
  }
 
  /** finds the next scope to enter.  the most outer scope is found first */
  public static ActivityImpl findNextScope(ScopeImpl outerScopeElement, ActivityImpl destination) {
    ActivityImpl nextScope = destination;
    while( (nextScope.getParent() instanceof ActivityImpl)
           && (nextScope.getParent() != outerScopeElement)
         ) {
      nextScope = (ActivityImpl) nextScope.getParent();
    }
    return nextScope;
  }
}

4.6 AtomicOperationTransitionCreateScope

public class AtomicOperationTransitionCreateScope implements AtomicOperation {
 
  private static Logger log = LoggerFactory.getLogger(AtomicOperationTransitionCreateScope.class);
 
  public boolean isAsync(InterpretableExecution execution) {
    ActivityImpl activity = (ActivityImpl) execution.getActivity();
    return activity.isAsync();
  }
 
  public void execute(InterpretableExecution execution) {
    InterpretableExecution propagatingExecution = null;
    ActivityImpl activity = (ActivityImpl) execution.getActivity();
    if (activity.isScope()) {
      //为作用域活动创建一个新的执行实例，是该原子操作的主要目的
      propagatingExecution = (InterpretableExecution) execution.createExecution();
      propagatingExecution.setActivity(activity);
      propagatingExecution.setTransition(execution.getTransition());
      execution.setTransition(null);
      execution.setActivity(null);
      execution.setActive(false);
      log.debug("create scope: parent {} continues as execution {}", execution, propagatingExecution);
      //这里是另外一个重点。在一个流程实例初始化的时候，会对当前流程所处的作用域对象。
      //可能是流程定义或者是作用域活动进行处理，具体表现是为该作用域对象上的定时事件，
      //消息事件，信号事件执行注册动作。分别是放入定时调度器，在数据库新增事件订阅
      propagatingExecution.initialize();
 
    } else {
      propagatingExecution = execution;
    }
 
    propagatingExecution.performOperation(AtomicOperation.TRANSITION_NOTIFY_LISTENER_START);
  }
}