工作流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>