工作流引擎Activiti使用進階！詳細解析工作流框架中高級功能的使用示例

Activiti高級功能簡介

• Activit的高級用例,會超越BPMN 2.0流程的范疇,使用Activiti高級功能需要有Activiti開發的明確目標和足夠的Activiti開發經驗

監聽流程解析

• bpmn 2.0 xml文件需要被解析為Activiti內部模型,然后才能在Activiti引擎中運行.解析過程發生在發布流程或在內存中找不到對應流程的時候,這時會從數據庫查詢對應的xml
• 對於每個流程 ,BpmnParser類都會創建一個新的BpmnParse實例.這個實例會作為解析過程中的容器來使用
• 解析過程:
  • 對於每個BPMN 2.0元素,引擎中都會有一個對應的org.activiti.engine.parse.BpmnParseHandler實例
  • 解析器會保存一個BPMN 2.0元素與BpmnParseHandler實例的映射
  • 默認Activiti使用BpmnParseHandler來處理所有支持的元素
  • 同時也使用BpmnParseHandler來提供執行監聽器,以支持流程歷史
• 可以向Activiti引擎中添加自定義的org.activiti.engine.parse.BpmnParseHandler實例
• 經常使用的用例是把執行監聽器添加到對應的環節,來處理一些事件隊列.Activiti在內部就是這樣進行歷史處理的
• 要想添加這樣的自定義處理器,需要為Activit增加配置:

<property name="preBpmnParseHandlers">
  <list>
    <bean class="org.activiti.parsing.MyFirstBpmnParseHandler" />
  </list>
</property>

<property name="postBpmnParseHandlers">
  <list>
    <bean class="org.activiti.parsing.MySecondBpmnParseHandler" />
    <bean class="org.activiti.parsing.MyThirdBpmnParseHandler" />
  </list>
</property>

• 當自定義處理器內部邏輯對處理順序有要求時需要考慮:
  • 配置到preBpmnParseHandlers的BpmnParseHandler實例會添加在默認處理器的前面
  • 配置到postBpmnParseHandlers的BpmnParseHandler實例會添加在默認處理器的后面
• 接口- org.activiti.engine.parse.BpmnParseHandler:

public interface BpmnParseHandler {

  Collection<Class>? extends BaseElement>> getHandledTypes();

  void parse(BpmnParse bpmnParse, BaseElement element);

}

• 在BpmnParseHandler接口中:
  • getHandledTypes()方法會翻譯這個解析器處理的所有類型的集合,這些都是BaseElement的子類,返回集合的泛型限制也說明了這一點
  • 也可以繼承AbstractBpmnParseHandler類並重寫getHandledType()方法,這樣就只需要返回一個類型,而不是一個集合
  • 這個類也包含需要默認解析處理器所需要的方法
  • BpmnParseHandler實例只有在解析器訪問到這個方法返回的類型時才會被調用
• 示例:
  • 當BPMN 2.0 xml包含process元素時,就會執行executeParse方法中的邏輯
  • 這是一個已經完成類型轉換的方法,替換BpmnParseHandler接口中的parse方法

public class TestBPMNParseHandler extends AbstractBpmnParseHandler<Process> {

  protected Class<? extends BaseElement> getHandledType() {
    return Process.class;
  }

  protected void executeParse(BpmnParse bpmnParse, Process element) {
     ..
  }

}

• 注意:
  • 在編寫自定義解析處理器時,不要使用任何解析BPMN 2.0結構的內部類,這會導致出現問題很難定義
  • 應該實現BpmnParseHandler接口或集成內部抽象類 org.activiti.engine.impl.bpmn.parser.handler.AbstractBpmnParseHandler
  • 也可以替換默認的BpmnParseHandler實例,把解析BPMN 2.0元素替換為解析Activiti內部模型:

<property name="customDefaultBpmnParseHandlers">
  <list>
    ...
  </list>
</property>

• 示例: 將所有任務強制設置為異步

public class CustomUserTaskBpmnParseHandler extends ServiceTaskParseHandler {

  protected void executeParse(BpmnParse bpmnParse, ServiceTask serviceTask) {

    // Do the regular stuff
    super.executeParse(bpmnParse, serviceTask);

    // Make always async
    ActivityImpl activity = findActivity(bpmnParse, serviceTask.getId());
    activity.setAsync(true);
  }

}

支持高並發的UUID的ID生成器

• 在高並發的場景中,默認的ID生成器可能因為無法很快的獲取新ID區域而導致異常
• 所有流程引擎都有一個ID生成器,默認的ID生成器會在數據庫划取一塊ID范圍,其余引擎不能使用相同范圍的ID
• 在引擎運行期間,當默認的ID生成器發現已經越過ID范圍時,就會啟動一個新事務來獲得新范圍.在極限的情況下,高負載會導致問題
• 對於大部分情況,默認ID生成已經足夠:
  • 默認的org.activiti.engine.impl.db.DbIdGenerator有一個idBlockSize屬性,可以配置獲取ID范圍的大小,這樣就可以改變獲取ID的行為
  • 另一個可以選用的默認ID生成器是org.activiti.engine.impl.persistence.StrongUuidGenerator:
    • 會在本地生成一個唯一的UUID作為所有實體的標識
    • 因為生成UUID不需要訪問數據庫,所以在高並發環境下的表現比較好
• 默認ID生成器的性能依賴於運行硬件
• 將UUID生成器配置到Activiti:

<property name="idGenerator">
    <bean class="org.activiti.engine.impl.persistence.StrongUuidGenerator" />
</property>

• 使用UUID生成器需要添加依賴:

 <dependency>
    <groupId>com.fasterxml.uuid</groupId>
    <artifactId>java-uuid-generator</artifactId>
    <version>3.1.3</version>
</dependency>

多租戶

• 多租戶:
  • 通常是在軟件需要作為多個不同組織服務時產生的概念
  • 關鍵是數據分片,組織不能看到其余組織的數據
  • 在這種場景下,組織,部門,小組就叫做租戶
• 多租戶和安裝多個實例是從基本上不同的:
  • 多租戶是一個Activiti流程引擎實例為每個組織分別運行,對應不同的數據表
  • 安裝多個Activiti流程引擎實例時,雖然Activiti是輕量級的,運行流程引擎不會消耗很多資源,但是增加了復雜性,並需要更多維護工作.然而對於一些場景,也是正確的解決方案
• Activiti的多租戶主要圍繞着數據分片來實現:
  • Activiti沒有強行校驗多租戶的規則,即Activiti不會校驗查詢和使用數據時用戶是否使用了正確的租戶
  • 校驗由Activiti引擎的調用者層負責完成
  • Activiti只確認租戶信息會被保存,並在查詢流程數據時會被用到
• 在向Activiti流程引擎發布流程定義時,需要傳遞一個租戶標識.是一個字符串,限制在256字符內,作為租戶的唯一標識

 repositoryService.createDeployment()
            .addClassPathResource(...)
            .tenantId("myTenantId")
            .deploy();

• 通過部署傳遞租戶Id有以下作用:
  • 所有包含在部署中的流程定義都會繼承部署的tenantId
  • 所有從這些流程定義發起的流程實例,都會繼承流程定義的tenantId
  • 所有流程實例運行階段創建的任務都會繼承流程實例的tenantId.單獨運行的task也可以包含tenantId
  • 所有流程實例運行階段創建的分支都會繼承流程實例的tenantId
  • 在流程本身或通過API觸發一個信號拋出事件可以通過tenantId實現.信號只會在租戶環境下執行:如果有多個信號捕獲事件,並且名字相同,實際只有正確的tenantId下的事件會被調用
  • 所有作業(定時器,異步調用)會集成tenantId,或者來自流程定義(比如定時開始事件),或流程實例(運行期創建的作業,比如異步調用).這樣其實潛在的可以支持為一些租戶指定不同優先級的自定義jobExecutor
  • 所有歷史實體(歷史流程實例,任務和節點)會從對應的運行狀態集成tenantId
  • 作為單獨的一部分,model也可以設置tenantId.這里的model用來存儲Activiti modeler設計的bpmn 2.0模型
• 為了確保流程數據使用tenantId,所有的查詢API都可以通過tenantId進行查詢,可以使用其他的實體的對應查詢實現替換:

runtimeService.createProcessInstanceQuery()
    .processInstanceTenantId("myTenantId")
    .processDefinitionKey("myProcessDefinitionKey")
    .variableValueEquals("myVar", "someValue")
    .list()

查詢API也允許對tenantId使用like語法， 也可以過濾未設置tenantId的實體

• 重要注意點:
  • 因為數據庫的限制,特別是處理null的唯一校驗.默認表示未設置租戶的tenantId的值是空字符串
  • 流程定義key,流程定義version,tenantId的組合應該是唯一的,這個有數據庫約束校驗這個規則
  • 要注意tenantId不應設置為null,會影響一些數據庫Oracle的查詢,會把空字符串當做null處理
  • 這也是為什么withoutTenantId查詢會檢查空字符串或null.這意味着相同的流程定義,即流程定義key相同可以部署到不同的租戶下,可以擁有各自的版本.當不使用租戶時也不會影響使用
• 這些限制不會影響Activiti在集群環境下運行
• 可以通過調用repositoryService的changeDeploymentTenantId(String deploymentId, String newTenantId) 修改tenantId. 會修改之前繼承的所有tenantId. 當需要從非多租戶環境向多租戶環境下切換時,會非常實用

執行自定義SQL

• Activiti API允許使用高級API操作數據庫:
  • 在查詢數據方面,查詢API和Native Query API是非常強大的
  • 但是對於某些情況,不夠輕便
  • 使用完全自定義的SQL語句:select, insert, update和delete.可以執行在Activiti的數據存儲之上,但是完全又可以配置在流程引擎中:比如使用事務
• 為了使用自定義SQL,Activiti引擎使用MyBatis框架的功能:
  • 因此使用自定義SQL的第一件事,要創建MyBatis映射類
  • 假設不需要全部的任務數據,只需要其中的一小部分.可以使用Mapper實現:

public interface MyTestMapper {

    @Select("SELECT ID_ as id, NAME_ as name, CREATE_TIME_ as createTime FROM ACT_RU_TASK")
    List<Map<String, Object>> selectTasks();

}

• Mapper需要設置到流程引擎配置中:

...
<property name="customMybatisMappers">
  <set>
    <value>org.activiti.standalone.cfg.MyTestMapper</value>
  </set>
</property>
...

• 這個Mapper是一個接口:
  • MyBatis框架會在運行階段為這個接口創建一個實例
  • 返回值是沒有類型的,是一個map的list,和對應的行列對應
  • 如果需要也可以使用MyBatis映射
• 執行上面的查詢:
  • 可以使用managementService.executeCustomSql方法
  • 這個方法需要一個CustomSqlExecution實體
  • 這個實體類是一個封裝類,隱藏了引擎的內部實現所需執行的信息
  • 但是由於Java泛型,查詢返回的結果可讀性差
• 示例:
  • mapper類和返回類型類
  • 簡單調用mapper方法 並返回結果

CustomSqlExecution<MyTestMapper, List<Map<String, Object>>> customSqlExecution =
          new AbstractCustomSqlExecution<MyTestMapper, List<Map<String, Object>>>(MyTestMapper.class) {

  public List<Map<String, Object>> execute(MyTestMapper customMapper) {
    return customMapper.selectTasks();
  }

};

List<Map<String, Object>> results = managementService.executeCustomSql(customSqlExecution);

list中的Map只包含id,name和create time, 不是全部的任務對象

• 可以通過這樣的方式執行任意SQL:

 @Select({
        "SELECT task.ID_ as taskId, variable.LONG_ as variableValue FROM ACT_RU_VARIABLE variable",
        "inner join ACT_RU_TASK task on variable.TASK_ID_ = task.ID_",
        "where variable.NAME_ = #{variableName}"
    })
    List<Map<String, Object>> selectTaskWithSpecificVariable(String variableName);

使用這種方法,任務表會與變量表關聯.只會獲得對應名稱的變量,任務Id和對應的數值會被返回

使用ProcessEngineConfigurator實現流程引擎配置

• 可以使用ProcessEngineConfigurator實現一種高級的擴展流程引擎的配置:
  • 創建一個org.activiti.engine.cfg.ProcessEngineConfigurator接口的實現
  • 注入到流程引擎配置里

<bean id="processEngineConfiguration" class="...SomeProcessEngineConfigurationClass">

    ...

    <property name="configurators">
        <list>
            <bean class="com.mycompany.MyConfigurator">
                ...
            </bean>
        </list>
    </property>

    ...

</bean>

• 實現ProcessEngineConfigurator接口需要實現兩個方法:
  • configure: 將ProcessEngineConfiguration作為參數,可以通過這種方法添加自定義配置,這個方法在流程創建之前,在所有默認配置執行之前保證調用到
  • getPriority: 如果一些configurator存在依賴項的時候,允許對configurator進行排序
• configurator實例:
  • LDAP集成:
    • 這個configurator用來替換默認的user和group管理器類,使用處理LDAP用戶存儲的類
    • 基本上一個configurator允許很大程度上修改或增強流程引擎,對高級的場景非常有用
  • 使用自定義的版本替換流程定義緩存, 如下:

public class ProcessDefinitionCacheConfigurator extends AbstractProcessEngineConfigurator {

    public void configure(ProcessEngineConfigurationImpl processEngineConfiguration) {
            MyCache myCache = new MyCache();
            processEngineConfiguration.setProcessDefinitionCache(enterpriseProcessDefinitionCache);
    }

}

• 流程引擎配置器也可以通過ServiceLoader自動從classpath中加載:
  • 放在jar中的configurator實現必須放在classpath下
  • 並在jar的META-INF/services目錄下包含一個org.activiti.engine.cfg.ProcessEngineConfigurator文件
  • 文件的內容是自定義實現的全類名
  • 當流程引擎啟動時,日志會顯示找到了哪些configurator

INFO  org.activiti.engine.impl.cfg.ProcessEngineConfigurationImpl  - Found 1 auto-discoverable Process Engine Configurators
INFO  org.activiti.engine.impl.cfg.ProcessEngineConfigurationImpl  - Found 1 Process Engine Configurators in total:
INFO  org.activiti.engine.impl.cfg.ProcessEngineConfigurationImpl  - class org.activiti.MyCustomConfigurator

• 這種ServiceLoader的方式在某些環境下可能無法正常運行.使用ProcessEngineConfiguration的enableConfiguratorServiceLoader屬性來禁用這個功能,這個屬性的默認值為true

啟動安全BPMN 2.0xml

• 大多數情況下,BPMN 2.0流程發布到Activiti引擎是在嚴格的控制下的
• 然而在某些情況下,可能需要把比較隨意的BPMN 2.0 xml上傳到引擎,這時就要要考慮惡意用戶會攻擊服務器
• 為了避免BPMN 2.0xml引擎服務器受到攻擊,可以在引擎中設置enableSafeBpmnXml:

<property name="enableSafeBpmnXml" value="true"/>

• 默認這個功能沒有開啟.因為這個功能需要使用StaxSource類
• 由於JDK6,JBoss使用的是舊版的xml解析實現,無法使用StaxSource類,所以不能啟用安全的BPMN 2.0xml
• 如果Activiti運行的平台支持安全的BPMN 2.0xml功能,建議打開

事件日志

• 在Activiti 5.16版本中,添加了事件日志機制:
  • 這種日志機制構建在通用目的下的Activiti引擎的事件機制,默認是禁用的
  • 目的是由引擎產生的事件會被捕獲,包含所有事件數據的map會被創建出來,並提供給org.activiti.engine.impl.event.logger.EventFlusher, 會把數據刷新到別的地方
  • 默認會使用一個簡單地基於數據庫的事件處理器或者叫作刷新器,會使用jackson把map轉換為JSON, 並保存到數據庫中的EventLogEntryEntity實體
  • 默認會創建數據庫日志表ACT_EVT_LOG. 如果沒有使用事件日志,可以刪除這個表
• 啟用數據庫日志:

processEngineConfiguration.setEnableDatabaseEventLogging(true);

或者在流程引擎運行階段:

databaseEventLogger = new EventLogger(processEngineConfiguration.getClock());
runtimeService.addEventListener(databaseEventLogger);

• EventLogger類可以繼承:
  • 在需要使用自定義的數據日志時:
    • createEventFlusher() 方法需要返回一個org.activiti.engine.impl.event.logger.EventFlusher接口的實例
    • managementService.getEventLogEntries(startLogNr, size) 可以獲取Actviti的EventLogEntryEntity實例
• 可以使用大數據的NoSQL存儲: MongoDb,Elastic Search等等來存儲JSON。
• 使用的類是可插拔的: org.activiti.engine.impl.event.logger.EventLogger/EventFlusher和很多EventHandler類
• 可以切換成自定義應用場景: 不在數據庫中存儲JSON,而是放到隊列或大數據存儲中
• 注意:
  • 事件日志機制是Activiti傳統歷史管理器的附加品
  • 雖然所有數據都在數據庫表中,但是並沒有為查詢優化,不容易獲取
  • 真實的使用場景:
    • 審計跟蹤
    • 將事件日志數據放到大數據存儲中