工作流Activiti框架的事務和並發！流程引擎中異步和排他操作詳細解析

事務和並發

異步操作

• Activiti通過事務方式執行流程,可以根據需求定制
• Activiti處理事務:
  • 如果觸發了Activiti的操作(開始流程,完成任務,觸發流程繼續執行),activiti會推進流程,直到每個分支都進入等待狀態
  • 抽象的說,會從流程圖執行深度優先搜索,如果每個分支都遇到等待狀態,就會返回
  • 等待狀態是稍后需要執行任務,Activiti會把當前狀態保存到數據庫中,然后等待下一次觸發
  • 觸發可能來自外部,比如用戶任務或接收到一個消息,也可能來自Activiti本身(定時器事件)
    
    流程包含用戶任務,服務任務和定時器事件
    完成用戶任務和校驗地址是在同一個工作單元中,兩者的成功和失敗是原子性的.意味着如果服務任務拋出異常,要回滾當前事務,這樣流程會退回到用戶任務,用戶任務就依然在數據庫里
    這就是activiti默認的行為.在(1)中應用或客戶端線程完成任務.這會執行服務,流程推進,直到遇到一個等待狀態,就是定時器(2),然后它會返回給調用者(3),並提交事務(如果事務是由Activiti開啟的)
• 有時需要自定義控制流程中事務的邊界,把業務邏輯包裹在一起.這就需要使用異步執行:
  
  完成了用戶任務,生成一個發票,把發票發送給客戶
  生成發票不在同一個工作單元內了.如果生成發票出錯不需要對用戶任務進行回滾
  Activiti實現的是完成用戶任務(1),提交事務,返回給調用者應用.然后在后台的線程中,異步執行生成發票.
  后台線程就是Activiti的Job執行器(一個線程池)周期對數據庫的Job進行掃描:當到達"generate invoice"任務,為Activiti創建一個稍后執行的Job"消息",並保存到數據庫.Job會被Job執行器獲取並執行.也會給本地Job執行器一個提醒,告訴有一個新Job,來增加性能
• 要想使用這個特性,要使用activiti:async="true" 擴展

<serviceTask id="service1" name="Generate Invoice" activiti:class="my.custom.Delegate" activiti:async="true" />

• activiti:async可以使用到以下BPMN任務類型中:
  • task
  • serviceTask,
  • scriptTask
  • businessRuleTask
  • sendTask
  • receiveTask
  • userTask
  • subProcess
  • callActivity
• 對於userTask,receiveTask和其他等待狀態,異步執行的作用是讓開始流程監聽器運行在一個單獨的線程或者事務中

排他任務

• 從Activiti 5.9開始 ,JobExecutor能保證同一個流程實例中的Job不會並發執行

排他任務的產生背景

• 一個並行網關,后面有三個服務任務,都設置為異步執行:
  • 這樣會添加三個job到數據庫里.一旦job進入數據庫,就可以被jobExecutor執行了.JobExecutor會獲取job,代理到工作線程的線程池中,在那里真正執行job
  • 就是說,使用異步執行,可以把任務分配給這個線程池(在集群環境，可能會使用多個線程池)
  • 產生一致性問題:
    • 考慮一下服務任務后的匯聚:當服務任務完成后,到達並發匯聚節點,需要決定是等待其他分支,還是繼續向下執行
    • 就是說,對每個到達並行匯聚的分支,都需要判斷是繼續還是等待其他分支的一個或多個分支
• 為什么會產生這樣的問題:
  • 因為服務任務配置成使用異步執行,可能相關的job都在同一時間被獲取,被JobExecutor分配給不同的工作線程執行
  • 結果是,三個單獨的服務執行使用的事務在到達並發匯聚時可能重疊:
    • 如果出現了這個問題,這些事務是互相不可見的,其他事務同時到達了相同的並發匯聚,假設都在等待其他分支
    • 然而,每個事務都假設在等待其他分支,所以沒有分支會越過並發匯聚繼續執行,流程實例會一直在等待狀態,無法繼續執行
• Activiti解決這個問題方式:
  • Activiti使用了樂觀鎖:
    • 當基於判斷的數據看起來不是最新的時候 (因為其他事務可能在提交之前進行了修改,會在每個事務里增加數據庫同一行的版本),這個時候,第一個提交的事務會成功,其他會因為樂觀鎖異常導致失敗
    • 這就解決了上面流程的問題:
      • 如果多個分支同步到達並行匯聚,會假設都在登錄,並增加父流程的版本號(流程實例)然后嘗試提交
      • 第一個分支會成功提交,其他分支會因為樂觀鎖導致失敗
      • 因為流程是被job觸發的,Activiti會嘗試在等待一段時間后嘗試執行同一個job,這段時間可以同步網關的狀態
• Activiti樂觀鎖是一個很好的解決方案嗎?
  • 樂觀鎖允許Activiti避免非一致性,確定流程不會"堵在匯聚網關": 或者所有分支都通過網關，或者數據庫中的job正在嘗試通過
  • 雖然這是一個對於持久性和一致性的完美解決方案,但對於上層來說不一定是期望的行為:
    • Activiti只會對同一個job重試估計次數(默認配置為3).之后,job還會在數據庫里,但是不會再重試了.意味着這個操作必須手工執行job的觸發
    • 如果job有非事務方面的效果,不會因為失敗的事務回滾:如果“預定演唱會門票”服務沒有與Activiti共享事務,重試job可能導致我們預定了過多門票
• 針對這些問題,在Activiti中推出了新的概念:排他job

排他Job

• 對於一個流程實例,排他任務不能同時執行兩個
• 考慮上面的流程:如果我們把服務任務申請為排他任務,JobExecutor會保證對應的job不會並發執行.
• 會保證無論什么時候獲取一個流程實例的排他任務,都會把同一個流程實例的其他任務都取出來,放在同一個工作線程中執行.保證job是順序執行的
• 從activiti 5.9開始,排他任務已經是默認配置.所以異步執行和定時器事件默認都是排他任務
• 如果你想把job設置為非排他,可以使用activiti:exclusive="false" 進行配置:

<serviceTask id="service" activiti:expression="${myService.performBooking(hotel, dates)}" activiti:async="true" activiti:exclusive="false" />

• 排他任務沒有性能問題:
  • 在高負載的情況下性能是個問題,高負載意味着JobExecutor的所有工作線程都一直在忙碌着
  • 使用排他任務,Activiti可以簡單的分布不同的負載.排他任務意味着同一個流程實例的異步執行會由相同的線程順序執行
  • 但是要考慮:如果有多個流程實例時.所有其他流程實例的job也會分配給其他線程同步執行
  • 意味着雖然Activiti不會同時執行一個流程實例的排他job,但是還會同步執行多個流程實例的異步執行
  • 通過一個總體的預測,在大多數場景下,排他任務都會讓單獨的實例運行的更迅速.而且,對於同一流程實例中的job,需要用到的數據也會利用執行的集群節點的緩存.如果任務沒有在同一個節點執行,數據就必須每次從數據庫重新讀取了

流程實例授權

• 默認所有人在部署的流程定義上啟動一個新流程實例,通過流程初始化授權功能定義的用戶和組,web客戶端可以限制哪些用戶可以啟動一個新流程實例
• Activiti引擎不會校驗授權定義: 這個功能只是為減輕web客戶端開發者實現校驗規則的難度
• 設置方法與用戶任務用戶分配類似,用戶或組可以使用activiti:potentialStarter標簽分配為流程的默認啟動者:

 <process id="potentialStarter">
     <extensionElements>
       <activiti:potentialStarter>
         <resourceAssignmentExpression>
           <formalExpression>group2, group(group3), user(user3)</formalExpression>
         </resourceAssignmentExpression>
       </activiti:potentialStarter>
     </extensionElements>
   <startEvent id="theStart"/>
   ...

user(user3)是直接引用了用戶user3,group(group3)是引用了組group3.如果沒顯示設置,默為群組

• 也可以使用process標簽的屬性activiti:candidateStarterUsers和activiti:candidateStarterGroups

 <process id="potentialStarter" activiti:candidateStarterUsers="user1, user2"
                                activiti:candidateStarterGroups="group1">
      ...
  

可以同時使用這兩個屬性

• 定義流程初始化授權后,開發者可以使用如下方法獲得授權定義.可以獲得給定的用戶能夠啟動哪些流程定義:

 processDefinitions = repositoryService.createProcessDefinitionQuery().startableByUser("userxxx").list();

• 可以獲得指定流程定義設置的潛在啟動者對應的IdentityLink:

 identityLinks = repositoryService.getIdentityLinksForProcessDefinition("processDefinitionId"); 

• 獲得可以啟動給定流程的用戶列表的示例:

  List<User> authorizedUsers =  identityService().createUserQuery().potentialStarter("processDefinitionId").list(); 

• 獲得可以啟動給定流程配置的群組的示例:

 List<Group> authorizedGroups =  identityService().createGroupQuery().potentialStarter("processDefinitionId").list();  

數據對象

• BPMN提供了一種功能,可以在流程定義或子流程中定義數據對象
• 根據BPMN規范,流程定義可以包含復雜XML結構,可以導入XSD定義
• 對於Activiti來說 ,作為Activiti首次支持的數據對象, 可以支持如下的XSD類型

	<dataObject id="dObj1" name="StringTest" itemSubjectRef="xsd:string"/>

	<dataObject id="dObj2" name="BooleanTest" itemSubjectRef="xsd:boolean"/>

	<dataObject id="dObj3" name="DateTest" itemSubjectRef="xsd:datetime"/>

	<dataObject id="dObj4" name="DoubleTest" itemSubjectRef="xsd:double"/>

	<dataObject id="dObj5" name="IntegerTest" itemSubjectRef="xsd:int"/>

	<dataObject id="dObj6" name="LongTest" itemSubjectRef="xsd:long"/>

• 數據對象定義會自動轉換為流程變量,名稱與name屬性對應
• 除了數據對象的定義之外,Activiti支持使用擴展元素來為這個變量賦予默認值:

<process id="dataObjectScope" name="Data Object Scope" isExecutable="true">
          <dataObject id="dObj123" name="StringTest123" itemSubjectRef="xsd:string">
            <extensionElements>
              <activiti:value>Testing123</activiti:value>
            </extensionElements>
          </dataObject>