Activiti6.0的探索纪要

前言

年初要作一个运维自动化平台，须要用到流程引擎，原本打算项目用golang写的，可是golang的流程引擎功能太简单实在是用不来，最后仍是选型java + activiti。到activiti官网一看，嘿出7.0告终果文档是刚写的还不全，咱们java仍是8的，7.0是匹配的java11，最终是问题太多只好放弃用activiti6.0了。

摸石头过河

虽然网上教程有很多，不过要真正跑起来着实不容易，有些内容好比参数的意义仍是看5.0的手册才弄明白的。

安装

解决依赖先跑起来

搭建环境：java8 + springboot2.1.3 + activiti6.0 + mysql

pom依赖：

<dependency>
    <groupId>org.activiti</groupId>
    <artifactId>activiti-spring-boot-starter-basic</artifactId>
	<version>${activiti.version}</version>
</dependency>
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>8.0.15</version>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-jdbc</artifactId>
    <version>2.1.0.RELEASE</version>
</dependency>
复制代码

配置数据库：

spring.datasource.url=jdbc:mysql://127.0.0.1:3306/activiti?nullCatalogMeansCurrent=true&serverTimezone=Asia/Shanghai
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
spring.datasource.username=root
spring.datasource.password=root
复制代码

而后启动，恭喜你，应用没起来，会出现以下报错：

Caused by: java.io.FileNotFoundException: class path resource [org/springframework/security/config/annotation/authentication/configurers/GlobalAuthenticationConfigurerAdapter.class] cannot be opened because it does not exist:

这个缘由是activiti6.0开发的时候springboot2.0还没出来以至年久失修，文件路径不对了，解决方案：

启动类前排除掉SecurityAutoConfiguration

@SpringBootApplication(exclude = SecurityAutoConfiguration.class)

再次启动，若是你碰到建立数据库表却找不到表的问题，两种解决方法：

• 手动建立表，表的路径在activiti-engine-6.0.0.jar/org/activiti/db/create
• 数据库配置中加上nullCatalogMeansCurrent=true

这是由于在org/activiti/engine/impl/db/DbSqlSession中activiti使用databaseMetaData.getTables寻找库表是否存在，而dbSqlSessionFactory获取到的catalog为空由于mysql使用schema标识库名而不是catalog，致使mysql扫描全部的库来找表，一旦其余库中有同名表activiti就觉得找到了其实表并不存在。nullCatalogMeansCurrent的意义就在于让mysql默认当前库，在mysql-connector-java 5.x该参数默认为true，但在6.x以上默认为false。

再次启动，仍然没起来：

Caused by: java.io.FileNotFoundException: class path resource [processes/] cannot be resolved to URL because it does not exist

这是由于activiti会到resource/processes下面寻找流程文件，建立该目录。

配置画图插件

activiti提供了多种设计器来画流程图，我安装了IDEA的插件和eclipse的插件，比较了下仍是eclipse的好用，因此只介绍eclipse 插件的安装过程。

1. 下个eclipse
2. 下载插件 Help --> Install New Software -->Add：

Names : Activiti BPMN 2.0 designer
Location : https://www.activiti.org/designer/update/
复制代码

3. 下完了New一个diagram

4. 画流程图，一个标准的流程图有一个开始事件和一个结束事件，画完了保存为.bpmn文件或者.bpmn20.xml文件，并放到resource/processes目录下，一个xml示例文件以下：

<?xml version="1.0" encoding="UTF-8"?>
<definitions xmlns="http://www.omg.org/spec/BPMN/20100524/MODEL" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:activiti="http://activiti.org/bpmn" xmlns:bpmndi="http://www.omg.org/spec/BPMN/20100524/DI" xmlns:omgdc="http://www.omg.org/spec/DD/20100524/DC" xmlns:omgdi="http://www.omg.org/spec/DD/20100524/DI" xmlns:tns="http://www.activiti.org/test" typeLanguage="http://www.w3.org/2001/XMLSchema" expressionLanguage="http://www.w3.org/1999/XPath" targetNamespace="http://www.activiti.org/test" id="m1551347612734" name="">
  <process id="machine_expansion" name="Machine Expansion" isExecutable="true" isClosed="false" processType="None">
    <startEvent id="start" name="开始"></startEvent>
    ...
复制代码

运行

使用runtimeservice建立一个createProcessInstanceBuilder，设置processDefinitionKey为xml中的process id，而后start就能够了。

@Autowired
    private RuntimeService runtimeService;
    @Override
    public String start(BaseParamDTO baseParamDTO) {
        String currentUserName = CurrentUserUtil.getCurrentUserName();
        identityService.setAuthenticatedUserId(currentUserName);
        ExpansionParamDTO expansionParamDTO = (ExpansionParamDTO) baseParamDTO.getParams();
        ProcessInstance machineExpansion = runtimeService.createProcessInstanceBuilder().processDefinitionKey(WorksheetTypeEnum.EXPANSION.getActivitiDefineName())
                .businessKey(expansionParamDTO.getAppName()).name(WorksheetTypeEnum.EXPANSION.name()).start();
        baseParamDTO.setId(machineExpansion.getProcessInstanceId());
        submit(baseParamDTO);
        return machineExpansion.getProcessInstanceId();
    }
复制代码

为了设置流程的startuser，必需要使用identityService.setAuthenticatedUserId设置，由于process获取的是当前线程的用户。

数据库表

要熟悉一个项目，最简单的方式就是看他的数据库表是怎么设计的，表里面都存了什么东西，activiti数据库中一共有28张表

• ACT_GE_*：通用表，通常不用关注，若是其余表中出现二进制的数据id，好比抛错，就到act_ge_bytearray中查询
• ACT_HI_*：历史信息，activiti历史信息存储时分4个级别，能够用spring.activiti.historyLevel配置，对应HistoryService
  • none：什么也不保存
  • activity：保存全部的流程实例、任务、活动信息；
  • audit：默认级别，保存全部的流程实例、任务、活动、表单属性；
  • full：流程过程当中所有保存，比audit多详情和变量；
• ACT_ID_*：用户认证信息，对应IdentityService；
• ACT_RE_*：流程中的部署和定义，对应RepositoryService；
• ACT_RU_*：运行时信息，流程级别相关的表对应RuntimeService，task级别表对应TaskService，job级别表对应ManagementService。

表名	描述
act_evt_log	事件日志表
act_ge_bytearray	二进制数据表,好比抛错
act_ge_property	activiti属性表
act_hi_actinst	历史的流程活动实例
act_hi_attachment	历史的流程附件
act_hi_comment	历史的描述信息
act_hi_detail	历史的流程运行中的细节信息
act_hi_identitylink	历史的用户关系信息
act_hi_procinst	历史的流程实例
act_hi_taskinst	历史的usertask实例
act_hi_varinst	历史的流程参数
act_id_group	用户组
act_id_info	用户详情信息
act_id_membership	用户和用户组关联关系
act_id_user	用户信息
act_procdef_info	流程定义扩展表
act_re_deployment	流程部署信息
act_re_model	流程模型
act_re_procdef	已部署过的流程
act_ru_deadletter_job	运行时中死掉的job
act_ru_event_subscr	运行时的事件监听
act_ru_execution	运行时的流程执行实例
act_ru_identitylink	运行时的用户关系信息
act_ru_job	运行时的job
act_ru_suspended_job	运行时暂停的job
act_ru_task	运行时的usertask实例
act_ru_timer_job	运行时的定时器job
act_ru_variable	运行时的参数

描述东西要举例

基础定义

在activiti中有几个基础定义：

• process：一个流程实例，好比定义了一个审批流程bpmn，如今run起来了，这就是一个流程实例。
• exection：流程执行实例，一个流程中有不少步骤，每个步骤都会触发一个流程执行实例。
• task：专门指usertask，当建立了usertask时，activiti会在exection中建立一个task。
• job：执行步骤时，每一个exection会生成job给jobExecutor去执行，若是是通常job就在act_ru_job中，定时任务就在act_ru_timer_job中，job暂停了就会进入act_ru_suspended_job，跑失败了就到act_ru_deadletter_job。
• activity：这个词在历史表中出现，理解上和exection差很少，区别在于exection是一个执行实例，一个执行实例能够执行多个activity，而activity是指流程中的一个步骤，会对应到bpmn中某一步。

几种经常使用的task的区别

eclipse设计器总共有8种task，目前我只用到其中的4种

用户任务（User Task）

用户任务用来设置须要人操做才能完成的任务如审批，执行到usertask时，流程会卡住，只有用户人工触发了流程才会继续。示例以下，这里没有在画bpmn时写死用户，而是在流程中经过taskService设置assigness，usertask完成只能是complete，后续的流程能够用参数和排他网关控制。

@Override
    public void submit(BaseParamDTO baseParamDTO) {
        String currentUserName = CurrentUserUtil.getCurrentUserName();
        ExpansionParamDTO expansionParamDTO = (ExpansionParamDTO) baseParamDTO.getParams();
        Task submit = taskService.createTaskQuery().processInstanceId(baseParamDTO.getId()).taskDefinitionKey(ExpansionDisplayEnum.SUBMIT.getFlow()).singleResult();
        taskService.setAssignee(submit.getId(), currentUserName);
        taskService.setVariable(submit.getId(), INPUT_PARAM, expansionParamDTO);
        taskService.setVariable(submit.getId(), INPUT_PRIORITY, baseParamDTO.getPriority());
        taskService.setPriority(submit.getId(), baseParamDTO.getPriority());
        if (StringUtils.isNotBlank(baseParamDTO.getComment())){
            taskService.addComment(submit.getId(), null, baseParamDTO.getComment());
        }
        taskService.complete(submit.getId());
    }
复制代码

用户任务涉及到的几个概念：

• assigness：该任务的执行人，谁执行的就写谁
• candidateUsers：该任务的候选人，须要哪些人执行的就写谁
• candidateGroups：该任务的候选组，须要哪一个组的人执行的就写哪些组
• comment：备注，这个信息会写到act_hi_comment表中，只能和usertask或者流程相关联

脚本任务（Script Task）

写一段脚本，执行到该步骤时自动执行脚本。（没用过）

服务任务（Service Task）

最经常使用的task，执行到该步骤时自动执行对应的java类

@Component
public class ApplyMachineTask implements JavaDelegate {

    public static final Logger LOGGER = LoggerFactory.getLogger(ApplyMachineTask.class);

    public static final String MACHINE_APPLY_ID = "applyId";
    public static final String MACHINE_APPLY_ID_LOCAL = "localApplyId";
    public static final String MACHINE_APPLY_CHECK = "applyCheckCount";

    @Autowired
    private MachineService machineService;

    @Override
    public void execute(DelegateExecution delegateExecution){
        ExpansionParamDTO param = delegateExecution.getVariable(INPUT_PARAM, ExpansionParamDTO.class);
        LOGGER.debug("start apply machine, param {}", param);
        param.setWorksheetId(delegateExecution.getProcessInstanceId());
        ApiResponseDTO<String> stringApiResponseDTO = machineService.applyMachine(param);
        if (stringApiResponseDTO.isSuccess()){
            delegateExecution.setVariable(MACHINE_APPLY_ID, stringApiResponseDTO.getBody());
            delegateExecution.setVariableLocal(MACHINE_APPLY_ID_LOCAL, stringApiResponseDTO.getBody());
            delegateExecution.setVariable(MACHINE_APPLY_CHECK, 0);
        } else {
            delegateExecution.setVariableLocal(ERROR_LOCAL, stringApiResponseDTO.getError());
            throw new BpmnError(ERROR_RETRY, stringApiResponseDTO.getError());
        }
    }
}
复制代码

这里说下参数设置，setVariable会设置流程全局变量，也就是流程中任意一个步骤都能获取到的参数，用来参数传递，而setVariableLocal设置的是局部变量，只有当前步骤才能获取的参数，局部变量的用处在于能够记录流程执行到该步骤时的状态，参数涉及到覆盖问题，因此不要出现全局和局部参数同名的现象（子流程的除外）。

业务规则任务（BusinessRule Task）

业务规则用户用来同步执行一个或多个规则。activiti使用drools规则引擎执行业务规则。（没用过）

邮件任务（Mail Task）

自动邮件任务，它能够发送邮件给一个或多个参与者。（没用过）

手工任务（Manual Task）

手工任务定义了BPM引擎外部的任务。 用来表示工做须要某人完成，而引擎不须要知道，也没有对应的系统和UI接口。 对于引擎，手工任务是直接经过的活动， 流程到达它以后会自动向下执行。（没用过）

接收任务（Receive Task）

接收任务是一个简单任务，它会等待对应消息的到达。流程到达该步骤后会一直等待，直到消息触发。以下图所示，在一个接收任务上挂了一个定时边界事件，定时边界事件每隔一段时间触发服务任务检查状态，当状态经过时给接收任务发消息，接收任务完成，定时边界网关中止。

定时边界网关配置

消息触发：

Execution receiveTask = runtimeService.createExecutionQuery().processInstanceId(execution.getProcessInstanceId())
    .activityId(ExpansionDisplayEnum.WAIT_INIT.getFlow()).singleResult();
runtimeService.trigger(receiveTask.getId());
复制代码

调用子流程（CallActivity）

activity有两种子流程，一种是CallActivity，一种是subprocess，区别在于CallActivity是调用主流程外部的一个流程，运行时会生成新的processId，做用域彻底独立，一般用来复用其余流程或者须要独立processId的状况；而subprocess是嵌套子流程，他并不会生成新的processId，可是定义了独立的事件域，一般用来捕捉同一类事件。

1. 画一个子流程，这里不能选async，选了就跑不起来了，个人理解是调用CallActivity和执行新的子流程是不一样的线程执行的，调用CallActivity只是为了生成建立子流程的job，而子流程是否异步执行是在multi instance页面配置的。

   

2. main config页面配置要调用的子流程的process id和须要传的参数，这里参数的意思是主流程的参数名privilege传过去后叫参数名param

   

3. multi instance多实例页配置，sequential就是配置是同步仍是异步的地方，loop 能够直接指定实例个数，这里经过传参解决，定义collection表示会有一个列表名字叫privilegeList的全局参数，列表中每一个参数实例的名字叫privilege，completion condition设置完成条件。当建立了多实例后activiti会自动建立多个参数：

• nrOfInstances：实例总数
• nrOfActiveInstances：当前活动的实例
• nrOfCompletedInstances：已经完成的实例

这里设置的条件是所有完成的时候主流程继续。

妈妈，我想用try catch

activiti有两种事件，一种是流程事件，放在流程中的，一种是边界事件，挂在某个步骤上因为流程运行间接触发的，而这两种事件又分为两类，一种是捕获一种是触发。

• 捕获（Catching）：当流程执行到事件， 它会等待被触发。
• 触发（Throwing）：当流程执行到事件， 会触发一个事件。

这里要说的就是边界错误捕获事件。

如图所示，把启动应用的全部步骤都放到了一个内嵌子流程subprocess中，而后在subprocess上挂载了一个错误边界捕获事件，当子流程中任意一个步骤抛错时捕获错误触发重试按钮，若是重启则从新执行整个子流程，若是跳过这流程继续。

错误边界事件中有一个errorCode参数，errorCode用来匹配捕获的错误：

• 若是没有设置errorRef，边界错误事件会捕获 全部错误事件，不管错误的errorCode是什么。
• 若是设置了errorRef，并引用了一个已存的错误， 边界事件就只捕获错误代码与之相同的错误。
• 若是设置了errorRef，可是BPMN 2.0中没有定义错误， errorRef就会当作errorCode使用 （和错误结束事件的用法相似）。

配置一个排他网关，在每条路径上写明条件

子流程每一个步骤中try catch住错误，并抛出

delegateExecution.setVariableLocal(ERROR_LOCAL,stringApiResponseDTO.getError());
throw new BpmnError(ERROR_RETRY, stringApiResponseDTO.getError());
复制代码

须要注意的是抛错中的错误文本好像获取不到，因此仍是要用局部变量来存一下。

如何作审批和驳回

由于我想记录全部的操做步骤，因此我把每一种操做都进行了分割，所以出现了下图所示的审批流程。

第一次提交任务是在流程开始时直接执行的，而后使用了监听器在审批步骤开始前设置审批候选人，若是当前用户是审批候选人直接跳过审批流程，审批是否经过是经过排他网关加上参数控制的。

审批usertask上设置一个监听器，用来设置任务的审批候选人

@Component
public class ApprovalTaskListener implements TaskListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(ApprovalTaskListener.class);

    @Autowired
    private RuntimeService runtimeService;

    @Autowired
    private ApprovalTask approvalTask;

    @Autowired
    private ApplicationService applicationService;

    @Autowired
    private TaskService taskService;

    @Override
    public void notify(DelegateTask delegateTask) {
        String processInstanceId = delegateTask.getProcessInstanceId();
        ProcessInstance processInstance = runtimeService.createProcessInstanceQuery().processInstanceId(processInstanceId).singleResult();
        String appName = processInstance.getBusinessKey();
        String owner = processInstance.getStartUserId();
        Integer priority = delegateTask.getVariable(INPUT_PRIORITY, Integer.class);
        taskService.setPriority(delegateTask.getId(), priority);
        ApiResponseDTO<ApplicationDTO> application = applicationService.getApplication(appName);
        if (application.isSuccess()){
            List<String> bizMaintainer = application.getBody().getBizMaintainer();
            delegateTask.addCandidateUsers(bizMaintainer);
            //申请人是审批人时直接处理
            if (bizMaintainer.contains(owner)){
                taskService.setAssignee(delegateTask.getId(), owner);
            }
        } else {
            LOGGER.error("get application info error, pId: {}, app: {}", processInstanceId, application);
            throw new RuntimeException(application.getError());
        }
    }
}
复制代码

在审批usertask外层设置exection的监听器approvalSkipListener，前面说过exection是先于usertask执行的，每一个usertask外层都有一个exection，因此这个监听器能够控制usertask的行为。

@Component
public class ApprovalSkipListener implements ExecutionListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(ApprovalSkipListener.class);

    @Autowired
    private RuntimeService runtimeService;

    @Autowired
    private ApplicationService applicationService;

    @Override
    public void notify(DelegateExecution execution) {
        ProcessInstance processInstance = runtimeService.createProcessInstanceQuery().processInstanceId(execution.getProcessInstanceId()).singleResult();
        String appName = processInstance.getBusinessKey();
        String owner = processInstance.getStartUserId();
        ApiResponseDTO<ApplicationDTO> application = applicationService.getApplication(appName);
        if (application.isSuccess()){
            List<String> bizMaintainer = application.getBody().getBizMaintainer();
            if (bizMaintainer.contains(owner)){
                execution.setVariable(SKIP_ENABLE,true);
                execution.setVariable(APPROVAL_PASS, true);
            }
        } else {
            LOGGER.error("get application info error, pId: {}, app: {}", processInstance.getId(), application);
            throw new RuntimeException(application.getError());
        }
    }
}
复制代码

设置跳过条件

若是驳回了，流程又会回到提交任务那，须要用户从新提交，数据库中会出现新的记录。

说回滚就回滚

activiti有个颇有意思的功能就是补偿，回滚就是用补偿触发事件和边界补偿捕获事件完成的。当补偿触发后，activiti会从后往前依次触发补偿捕获事件。

1. 画一个补偿触发事件

   

2. 画一个边界补偿捕获事件

   

   上图的意思是当流程走到静默期这个接收任务的时候会卡住并触发一个边界定时事件，若是定时到了就会走机器下线流程结束，若是定时没到却触发了静默期消息就会走补偿触发事件，触发应用启动，回滚成功。

问题不少慢慢解

任务怎么没记录

说到这个问题，就要说起异步任务：

• async：标识任务是否是异步执行的，若是为true表示任务不是jobExecutor直接执行，而是生成额外的job放到工做线程池由工做线程执行。

借用官方的图来简要说明一下。activiti是经过事务的方式执行流程，当流程开始后，activit会一直推动流程直到遇到等待状态，而后把当前状态存到数据库，并等待下一次触发。以下图中usertask是第一个等待状态，定时器是第二个等待状态，而完成usertask和执行服务任务在同一个工做单元中，同一个工做单元中的成功和失败是原子的。若是服务任务成功了数据库中看到的是完成的usertask和完成的服务任务，若是服务任务出错了就会致使activti回滚事务，这个时候就会回到最初的状态，现象就是数据库中依然只有未完成的usertask的记录，而看不到服务任务的记录。

若是我想看到服务任务的执行状态呢，解法就是设置async为true，将服务任务交给后台执行，后台的JobExecutor会周期性的扫描数据库的job提交给工做线程池。以下图所示，服务任务设置了async为true，这个时候有三个等待状态，第一个是usertask，第二个是服务任务，第三个是定时器。如今流程开始，usertask完成后会建立一个等待状态的服务任务job并把他保存到数据库，而后由工做线程池执行，若是服务任务job执行失败了，这个job仍然在数据库中。而邮件任务若是失败了则会和以前同样回滚到服务任务初始状态。

为啥job不跑了

这也是常常遇到的一个坑，目前尚未解决，这涉及到排他任务。

• exclusive：表示任务是否是排他任务，若是为true，工做线程会把同一个流程实例的排他任务一次所有拿出来顺序执行。

仍是借用官方的图，假设有以下场景，三个服务任务都是异步的，jobExecutor会把这3个job分给工做线程池来执行，若是他们同时到达并发汇聚网关那就会出现一致性问题，由于每一个分支都会检查其余分支是否到达就会出现一直等待的状况。

为了解决这个问题，activiti使用了乐观锁，当全部分支汇聚时他们会去更新流程实例的版本号，若是有分支更新失败则会被锁上一段时间而后重试。可是乐观锁的问题在于失败的job默认只重试三次并且job是非事务的会致使数据重复。所以又出现了排他任务，exclusive为true时保证activiti对于同一个流程实例不会同时执行两个排他任务，也就是说async和exclusive同时为true时并非真正的并行，而6.0 exclusive默认为true。

按道理只有async为true，exclusive为false时才会job锁住不跑了，可是事实并非这样，所以我怀疑是当多台机器同时获取job时致使的这个问题，还待后续排查。

后记

洋洋洒洒终于把activiti的总结写完了，不过是点到为止，还有不少功能没有尝试，bug也没查出来。革命还没有成功，同志仍需努力。