activiti流程運(yùn)行過程代碼走讀

1. 目的

本文將對流程在activiti中是怎么運(yùn)行的，任務(wù)是怎樣推動的進(jìn)行講解。

主要包括以下內(nèi)容：

  1. PVM中怎么表示流程、任務(wù)、連接線，它和activiti的Model怎樣轉(zhuǎn)換的。

  • 啟動流程實(shí)例，怎么從開始節(jié)點(diǎn)流轉(zhuǎn)到下一個節(jié)點(diǎn)。

  • 怎樣監(jiān)聽流程事件。

2. 主要的jar包和java類

pvm的實(shí)現(xiàn)在包org.activiti.engine.impl.pvm中，我通過源碼跟蹤、閱讀，分析出PVM有以下重要的類：

class 功能
ProcessDefinitionImpl 流程定義
ExecutionEntity 管理流程的運(yùn)行
ActivityImpl 流程節(jié)點(diǎn)的定義
AtomicOperation 流程運(yùn)行方法，如：
　AtomicOperationActivityStart(流程啟動)，
　AtomicOperationActivityEnd(流程結(jié)束)
ActivityBehavior 委托

2.1 在哪定義的

2.1.1 ProcessDefinitionImpl

ProcessDefinitionImpl是PVM對整個流程定義。通過ProcessDefinitionImpl.createProcessInstance可以得到流程實(shí)例管理接口PvmProcessInstance。ProcessDefinitionImpl的代碼片段如下：

  public class ProcessDefinitionImpl extends ScopeImpl implements PvmProcessDefinition {

    private static final long serialVersionUID = 1L;

    protected String name;
    protected String key;
    protected String description;
    protected ActivityImpl initial;
    protected Map<ActivityImpl, List<ActivityImpl>> initialActivityStacks = new HashMap<ActivityImpl, List<ActivityImpl>>();
    protected List<LaneSet> laneSets;
    protected ParticipantProcess participantProcess;

    public ProcessDefinitionImpl(String id) {
      super(id, null);
      processDefinition = this;
    }

    public PvmProcessInstance createProcessInstance() {
      if(initial == null) {
        throw new ActivitiException("Process '"+name+"' has no default start activity (e.g. none start event), hence you cannot use 'startProcessInstanceBy...' but have to start it using one of the modeled start events (e.g. message start events).");
      }
      return createProcessInstanceForInitial(initial);
    }

    ...  
  }

2.1.2 ExecutionEntity

ExecutionEntity實(shí)現(xiàn)了接口ActivityExecution，ActivityExecution是流程運(yùn)行管理接口。ExecutionEntity提供的功能如下：

  1. 啟動、結(jié)束、銷毀流程。
  • 對流程元素ActivityImpl的管理（添加刪除修改父節(jié)點(diǎn)、實(shí)例ID、任務(wù)等）。

ExecutionEntity里幾個重要的方法：

  1. initialize()里面初始化task、job、variable、event。

  • start()是流程啟動方法。

  • performOperation()是流程執(zhí)行方法，流程的推動都調(diào)用performOperation()。

2.1.3 ActivityImpl

ActivityImpl表示單個流程元素，比如任務(wù)、開始節(jié)點(diǎn)、連接線、網(wǎng)關(guān)等。它的屬性包括：variables、activityBehavior、incomingTransitions、outgoingTransitions以及元素界面位置等。ActivityImpl的類圖如下：

類圖

  • ActivityImpl是活動節(jié)點(diǎn)，在流程圖上對應(yīng)開始節(jié)點(diǎn)、結(jié)束節(jié)點(diǎn)、各種任務(wù)節(jié)點(diǎn)。

  • TransitionImpl是連接線。

  • ProcessDefinitionImpl是整個的定義，類圖中未畫。類ProcessElementImpl里面有ProcessDefinitionImpl屬性。

2.1.4 AtomicOperation

AtomicOperation是流程執(zhí)行的接口，比如流程啟動，流程結(jié)束，任務(wù)啟動，任務(wù)執(zhí)行，任務(wù)結(jié)束等。AtomicOperation中枚舉了所有流程執(zhí)行方法，AtomicOperation的代碼如下：


  public interface AtomicOperation {

    AtomicOperation PROCESS_START = new AtomicOperationProcessStart();
    AtomicOperation PROCESS_START_INITIAL = new AtomicOperationProcessStartInitial();
    AtomicOperation PROCESS_END = new AtomicOperationProcessEnd();
    AtomicOperation ACTIVITY_START = new AtomicOperationActivityStart();
    AtomicOperation ACTIVITY_EXECUTE = new AtomicOperationActivityExecute();
    AtomicOperation ACTIVITY_END = new AtomicOperationActivityEnd();
    AtomicOperation TRANSITION_NOTIFY_LISTENER_END = new AtomicOperationTransitionNotifyListenerEnd();
    AtomicOperation TRANSITION_DESTROY_SCOPE = new AtomicOperationTransitionDestroyScope();
    AtomicOperation TRANSITION_NOTIFY_LISTENER_TAKE = new AtomicOperationTransitionNotifyListenerTake();
    AtomicOperation TRANSITION_CREATE_SCOPE = new AtomicOperationTransitionCreateScope();
    AtomicOperation TRANSITION_NOTIFY_LISTENER_START = new AtomicOperationTransitionNotifyListenerStart();

    AtomicOperation DELETE_CASCADE = new AtomicOperationDeleteCascade();
    AtomicOperation DELETE_CASCADE_FIRE_ACTIVITY_END = new AtomicOperationDeleteCascadeFireActivityEnd();

    void execute(InterpretableExecution execution);

    boolean isAsync(InterpretableExecution execution);
  }

2.1.5 ActivityBehavior

ActivityBehavior是委托接口。AtomicOperation執(zhí)行過程中會調(diào)用ActivityBehavior。ActivityBehavior定義了execute方法，入?yún)㈩愋蜑?a href="#212-executionentity" target="_blank">ActivityExecution，ActivityBehavior的代碼如下：

public interface ActivityBehavior extends Serializable {
  void execute(ActivityExecution execution) throws Exception;
}

2.2 模塊之間調(diào)用順序

下面用RuntimeServiceImpl.startProcessInstanceById()啟動一個流程，說明PVM的調(diào)用順序。流程圖如下：

示例

2.2.1 從activiti調(diào)用PVM

RuntimeServiceImpl.startProcessInstanceById()通過命令模式調(diào)用了方法StartProcessInstanceCmd.execute()，StartProcessInstanceCmd實(shí)現(xiàn)了以下功能：

  1. 從數(shù)據(jù)庫獲取ProcessDefinitionEntity（ProcessDefinitionEntity是流程對應(yīng)的數(shù)據(jù)庫實(shí)體，同時繼承ProcessDefinitionImpl）。

  • 通過方法ProcessDefinitionEntity.createProcessInstance獲得ExecutionEntity。

  • 調(diào)用ExecutionEntity.start()啟動流程。

2.2.2 ExecutionEntity調(diào)用AtomicOperation

<a id="start"></a>
StartProcessInstanceCmd調(diào)用了ExecutionEntity.start()就完了，任務(wù)怎么生成的?？隙ㄔ?code>ExecutionEntity.start()里。ExecutionEntity.start()的代碼如下：

  public void start() {
    if(startingExecution == null && isProcessInstanceType()) {
      startingExecution = new StartingExecution(processDefinition.getInitial());
    }
    performOperation(AtomicOperation.PROCESS_START);
  }

ExecutionEntity.start()里面調(diào)用了performOperation，performOperation又對同步執(zhí)行和異步執(zhí)行進(jìn)行了拆分，這里只看同步執(zhí)行的代碼：

//  activiti里面到處都是`CommandContext`
protected void performOperationSync(AtomicOperation executionOperation) {
  Context
    .getCommandContext()
    .performOperation(executionOperation, this);
}

Context.getCommandContext().performOperation里面調(diào)用了executionOperation.execute()。Context里對線程之間的數(shù)據(jù)進(jìn)行了隔離，但performOperation對并發(fā)調(diào)用的判斷是多余的。一次部署過程需要調(diào)用10多次performOperation，很有必要看看代碼長什么樣，Context.getCommandContext().performOperation的代碼如下：

  public void performOperation(AtomicOperation executionOperation, InterpretableExecution execution) {
    nextOperations.add(executionOperation);
    if (nextOperations.size()==1) {
      try {
        Context.setExecutionContext(execution);
        while (!nextOperations.isEmpty()) {
          AtomicOperation currentOperation = nextOperations.removeFirst();
          if (log.isTraceEnabled()) {
            log.trace("AtomicOperation: {} on {}", currentOperation, this);
          }
          if (execution.getReplacedBy() == null) {
            currentOperation.execute(execution);
          } else {
            currentOperation.execute(execution.getReplacedBy());
          }
        }
      } finally {
        Context.removeExecutionContext();
      }
    }
  }

2.2.3 AtomicOperation執(zhí)行順序

繞了一大圈，ExecutionEntity.start()真正調(diào)用了AtomicOperation.PROCESS_START(AtomicOperation.PROCESS_START = new AtomicOperationProcessStart())。AtomicOperationProcessStart實(shí)現(xiàn)了以下功能：

  1. 觸發(fā)listener、event。

  • 設(shè)置流程運(yùn)行節(jié)點(diǎn)為開始節(jié)點(diǎn)。

  • 調(diào)用AtomicOperation.PROCESS_START_INITIAL

AtomicOperationProcessStart的代碼片段如下：

@Override
  protected void eventNotificationsCompleted(InterpretableExecution execution) {
    if(Context.getProcessEngineConfiguration() != null && Context.getProcessEngineConfiguration().getEventDispatcher().isEnabled()) {
      Map<String, Object> 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));
    }

    ProcessDefinitionImpl processDefinition = execution.getProcessDefinition();
    StartingExecution startingExecution = execution.getStartingExecution();
    List<ActivityImpl> initialActivityStack = processDefinition.getInitialActivityStack(startingExecution.getInitial());  
    execution.setActivity(initialActivityStack.get(0));   // 當(dāng)前流程執(zhí)行節(jié)點(diǎn)就通過execution.setActivity設(shè)置了    
    execution.performOperation(PROCESS_START_INITIAL);    //  當(dāng)前操作完成之后，進(jìn)行下個操作就調(diào)用execution.performOperation
  }

AtomicOperation.PROCESS_START_INITIAL里面又調(diào)用了AtomicOperation.ACTIVITY_EXECUTE，經(jīng)過多次調(diào)用不同的AtomicOperation接口，完成流程實(shí)例的創(chuàng)建。

對于示例流程圖，從開始節(jié)點(diǎn)到用戶任務(wù)的調(diào)用AtomicOperation順序如下：

PROCESS_START  
PROCESS_START_INITIAL  
ACTIVITY_EXECUTE  
TRANSITION_NOTIFY_LISTENER_END  
TRANSITION_DESTROY_SCOPE    
TRANSITION_NOTIFY_LISTENER_TAKE    
TRANSITION_CREATE_SCOPE    
TRANSITION_NOTIFY_LISTENER_START   
ACTIVITY_EXECUTE

2.2.4 Task節(jié)點(diǎn)多久持久化到數(shù)據(jù)庫的

用戶任務(wù)對應(yīng)的ActivityImpl上綁定了UserTaskActivityBehavior，UserTaskActivityBehavior中對任務(wù)數(shù)據(jù)進(jìn)行了持久化處理。

用戶任務(wù)執(zhí)行ACTIVITY_EXECUTE操作的時候，會觸發(fā)節(jié)點(diǎn)上綁定的Behavior。

3. 看后感

  1. 接口定義到處都是，對于后期擴(kuò)展是有好處的，關(guān)鍵代碼要找好久。

  • PVM的AtomicOperation很適合用命令模式，為什么要用委托的方式。

  • AtomicOperation的調(diào)用棧太深了。

  • CommandContext到處都是。

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容