Activiti 框架原理

本文基于一个简单的Demo流程介绍了Activiti框架启动、部署、运行过程。

Demo准备

流程图文件：

<?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" typeLanguage="http://www.w3.org/2001/XMLSchema" expressionLanguage="http://www.w3.org/1999/XPath" targetNamespace="http://www.activiti.org/test">
   <process id="hello" name="hello" isExecutable="true">
        <!-- 流程开始节点 -->
        <startEvent id="start" name="Start" ></startEvent>
        <!-- serviceTask：执行me.likeyao.activiti.demo.HelloWorld的execute方法，打印hello world -->
        <serviceTask id="helloworld" name="helloworld" activiti:class="me.likeyao.activiti.demo.HelloWorld"/>
        <!-- 流程结束节点 -->
        <endEvent id="end" name="End"></endEvent>
        <!-- 流程迁移线：开始节点到serviceTask节点 -->
        <sequenceFlow id="sid-1" sourceRef="start" targetRef="helloworld"></sequenceFlow>
        <!-- 流程迁移线：serviceTask节点到结束节点 -->
        <sequenceFlow id="sid-3" sourceRef="helloworld" targetRef="end"></sequenceFlow>
    </process>
</definitions>

流程图： demo流程

代码：

public class App {
    public static void main(String[] args) {
        //创建流程引擎
        ProcessEngine processEngine = ProcessEngines.getDefaultProcessEngine();
        //部署流程图
        processEngine.getRepositoryService().createDeployment().addClasspathResource("hello.bpmn20.xml").deploy();
        //发起流程
        processEngine.getRuntimeService().startProcessInstanceByKey("hello");
    }
}


public class HelloWorld implements JavaDelegate{
    public void execute(DelegateExecution execution) throws Exception {
        System.out.println("Hello world!");
    }
}

Demo实现的功能是发起一个流程，执行到流程的serviceTask节点时，打印Hello world！，然后流程结束。

源码版本：5.22.0

框架初始化

ProcessEngine类图

ProcessEngine

ProcessEngine是Activiti框架的门面，ProcessEngine本身不提供任何功能，通过getXXXService方法可以获取到对应的Service对象执行操作。Demo中涉及到的两个Service：

• RepositoryService：流程定义和流程部署相关功能。
• RuntimeService：流程实例相关功能（发起流程、获取流程实例变量）。

ProcessEngineConfiguration

ProcessEngineConfiguration负责Activiti框架的属性配置、初始化工作，初始化入口是buildProcessEngine方法，所有Activiti框架运行时需要用到的组件基本都在这里初始化：

public ProcessEngine buildProcessEngine() {
    init();
    return new ProcessEngineImpl(this);
}

protected void init() {
    initConfigurators();
    configuratorsBeforeInit();
    initProcessDiagramGenerator();
    initHistoryLevel();
    initExpressionManager();
    initDataSource();
    initVariableTypes();
    initBeans();
    initFormEngines();
    initFormTypes();
    initScriptingEngines();
    initClock();
    initBusinessCalendarManager();
    initCommandContextFactory();
    initTransactionContextFactory();
    initCommandExecutors();
    initServices();
    initIdGenerator();
    initDeployers();
    initJobHandlers();
    initJobExecutor();
    initAsyncExecutor();
    initTransactionFactory();
    initSqlSessionFactory();
    initSessionFactories();
    initJpa();
    initDelegateInterceptor();
    initEventHandlers();
    initFailedJobCommandFactory();
    initEventDispatcher();
    initProcessValidator();
    initDatabaseEventLogging();
    configuratorsAfterInit();
}

这里有一个扩展点：ProcessEngineConfigurator。

public interface ProcessEngineConfigurator {
    //组件初始化前
    void beforeInit(ProcessEngineConfigurationImpl processEngineConfiguration);
    //组件初始化后
    void configure(ProcessEngineConfigurationImpl processEngineConfiguration);
    //优先级
    int getPriority();
}

在init初始化方法中，initConfigurators方法通过ServiceLoader加载ProcessEngineConfigurator。随后在configuratorsBeforeInit和configuratorsAfterInit方法中分别调用ProcessEngineConfigurator的beforeInit和configure方法，使用户可以在ProcessEngineConfiguration初始化前后编程式的修改属性，替换Activiti默认组件。

流程部署

流程部署实现的功能是将xml格式的流程图，转化为Activiti框架运行时依赖的流程定义对象。

RepositoryService

RepositoryService类图

Demo中通过以下代码部署了一个流程：

processEngine.getRepositoryService().createDeployment().addClasspathResource("hello.bpmn20.xml").deploy();

createDeployment方法中创建了DeploymentBuilder对象，DeploymentBuilder对象负责读取指定路径的流程图xml文件的内容（byte数组），并缓存在DeploymentEntity对象中：

public DeploymentBuilder addInputStream(String resourceName, InputStream inputStream) {
    ...
    byte[] bytes = IoUtil.readInputStream(inputStream, resourceName);
    ResourceEntity resource = new ResourceEntity();
    resource.setName(resourceName);
    resource.setBytes(bytes);
    deployment.addResource(resource);
    return this;
}

最终DeploymentBuilder的deploy方法会调用RepositoryService的deploy方法，完成流程部署：

public Deployment deploy() {
    return repositoryService.deploy(this);
}

CommandExecutor

在RepositoryService的deploy方法中，使用了CommandExecutor对象：

public Deployment deploy(DeploymentBuilderImpl deploymentBuilder) {
    return commandExecutor.execute(new DeployCmd<Deployment>(deploymentBuilder));
}

CommandExecutor类图

在Activiti中，大部分操作都以Command模式实现，例如部署流程图的DeployCmd。CommandExecutor封装了一系列的CommandInterceptor，在内部形成CommandInterceptor链，在命令执行前后做了拦截。Activiti框架提供了一些

CommandInterceptor实现：

名称	作用
CommandContextInterceptor	用于生成命令执行的上下文（CommandContext）。
LogInterceptor	开启日志Debug级别后，打印日志。
JtaTransactionInterceptor	开启Jta事务

引入activiti-spring包，通过SpringTransactionInterceptor引入Spring的事务支持。

CommandExecutor在ProcessEngineConfigurationImpl的initCommandExecutors方法中初始化：

protected void initCommandExecutors() {
    initDefaultCommandConfig();
    initSchemaCommandConfig();
    initCommandInvoker();
    initCommandInterceptors();
    initCommandExecutor();
}

可以设置ProcessEngineConfigurationImpl的customPreCommandInterceptors和customPostCommandInterceptors属性，添加自定义的CommandInterceptor：

protected void initCommandInterceptors() {
    if (commandInterceptors==null) {
        commandInterceptors = new ArrayList<CommandInterceptor>();
        if (customPreCommandInterceptors!=null) {
            commandInterceptors.addAll(customPreCommandInterceptors);
        }
        commandInterceptors.addAll(getDefaultCommandInterceptors());
        if (customPostCommandInterceptors!=null) {
            commandInterceptors.addAll(customPostCommandInterceptors);
        }
        commandInterceptors.add(commandInvoker);
    }
}

这里的pre和post是指Activiti框架getDefaultCommandInterceptors()的前后。

CommandInvoker是CommandInterceptor链的最后一个对象，负责调用Command：

public class CommandInvoker extends AbstractCommandInterceptor {
    @Override
    public <T> T execute(CommandConfig config, Command<T> command) {
        return command.execute(Context.getCommandContext());
    }
}

CommandContext

CommandContext类图

CommandContext是Activit框架Command执行的上下文，主要包含各种SessionFactory：

sessionFactories = processEngineConfiguration.getSessionFactories();

SessionFactory负责生成Session，Session是Activiti操作持久化对象的统一接口：

名称	作用
ProcessDefinitionEntityManager	流程定义相关读写操作。
ExecutionEntityManager	流程实例相关读写操作。
DefaultHistoryManager	历史记录相关读写操作

CommandContext的生命周期

CommandConext在CommandContextInterceptor中创建，在finally代码块中销毁：

public <T> T execute(CommandConfig config, Command<T> command) {
    //首先尝试从线程上下文的栈中获取CommandContext
    CommandContext context = Context.getCommandContext();
    boolean contextReused = false;
    //什么时候创建新的CommandContext？
    //1、CommandConfig中指定了不复用CommandContext
    //2、当前线程上下文中不存在CommandConext
    //3、当前线程上下文中的CommandConext已经抛出异常
    if (!config.isContextReusePossible() || context == null || context.getException() != null) { 
        context = commandContextFactory.createCommandContext(command);      }  
    else {
        contextReused = true;
    }

    try {
        //将前面获取到的CommandContext入栈
        Context.setCommandContext(context);
        Context.setProcessEngineConfiguration(processEngineConfiguration);
        //执行下一个interceptor，在CommandInvoker中可以通过Context.getCommandContext()获取线程上下文中的CommandContext
        return next.execute(config, command);
    } catch (Exception e) {
        //记录异常信息
        context.exception(e);
    } finally {
        try {
            //如果CommandContext不可复用，用完直接关闭
            if (!contextReused) {
                context.close();
            }
        } finally {
            //出栈操作
            Context.removeCommandContext();
            Context.removeProcessEngineConfiguration();
            Context.removeBpmnOverrideContext();
        }
    }
    
    return null;
}

Activiti的框架可以在一个Command的执行过程中，调用另外一个Command，所以会出现是否需要复用CommandContext的选项，默认值为true。

流程的解析

在DeployCmd中，首先调用DeploymentEntityManager持久化存储DeploymentEntity对象：

commandContext.getDeploymentEntityManager().insertDeployment(deployment);

然后调用DeploymentManager部署流程（流程解析）：

commandContext.getProcessEngineConfiguration().getDeploymentManager().deploy(deployment, deploymentSettings);

DeploymentEntityManager

DeploymentEntityManager类图

DeploymentEntityManager的deploy方法中循环调用Deployer对象的deploy方法，Activiti默认的Deployer是BpmnDeployer。

另外DeploymentEntityManager中还缓存了解析好的流程定义对象和Bpmn模型对象。

Activiti持久化的是流程图xml文件，每次系统重新启动都要执行一次“deploy”操作，生成ProcessDefinitionEntity对象。

BpmnDeployer

BpmnDeployer类图

BpmnDeployer的deploy方法中包含几个操作（代码缩略版）：

public void deploy(DeploymentEntity deployment, Map<String, Object> deploymentSettings) {
    ...
    BpmnParse bpmnParse = bpmnParser.createParse().sourceInputStream(inputStream).setSourceSystemId(resourceName).deployment(deployment).name(resourceName);
    bpmnParse.execute();
    for (ProcessDefinitionEntity processDefinition: bpmnParse.getProcessDefinitions()) {
        if (deployment.isNew()) {
            ProcessDefinitionEntity latestProcessDefinition = ...
            if (latestProcessDefinition != null) {
                processDefinitionVersion = latestProcessDefinition.getVersion() + 1;
            }else{
                processDefinitionVersion = 1;
            }
            processDefinition.setId(idGenerator.getNextId());
            dbSqlSession.insert(processDefinition);
        }
        ...
    }
}

• 通过BpmnParser对象创建BpmnParse。
• 调用BpmnParse的execute方法，将inputStream中的流程图转化为ProcessDefinitionEntity。
• 持久化ProcessDefinitionEntity对象。

BpmnParse

BpmnParse类图

在BpmnParse的execute中完成了xml文件到ProcessDefinitionEntity对象的转化：

public BpmnParse execute() {
    //xml->bpmnModel
    bpmnModel = converter.convertToBpmnModel(streamSource, validateSchema, enableSafeBpmnXml, encoding);
    //bpmnModel-> ProcessDefinitionEntity
    transformProcessDefinitions();
}

protected void transformProcessDefinitions() {
    for (Process process : bpmnModel.getProcesses()) {
        bpmnParserHandlers.parseElement(this, process);
    }
}

在流程定义解析过程中，会涉及到两套模型：

• Bpmn模型（由BpmnXMLConverter完成转换）
• PVM模型（由BpmnParseHandlers完成转换）

Bpmn模型

Bpmn模型

PVM模型

PVM模型

Bpmn模型更偏向于xml节点的描述，PVM模型是运行时模型。Bpmn模型中的ServiceTask、StartEvent等会统一映射转换为PVM的ActivityImpl对象，ServiceTask和StartEvent等节点行为上的差别，体现在ActivityImpl对象持有的不同的ActivityBehavior上。

运行流程

创建流程实例

在demo中通过RuntimeService发起流程实例：

processEngine.getRuntimeService().startProcessInstanceByKey("hello");

在startProcessInstanceByKey方法中执行StartProcessInstanceCmd命令：

public class StartProcessInstanceCmd<T> implements Command<ProcessInstance>, Serializable {
    ...
    public ProcessInstance execute(CommandContext commandContext) {
        //获取流程定义
        ProcessDefinitionEntity processDefinition = ...
        //创建流程实例
        ExecutionEntity processInstance = processDefinition.createProcessInstance(businessKey);
        //开始流程
        processInstance.start();
        return processInstance;
    }
    ...
}

在StartProcessInstanceCmd方中通过流程定义ProcessDefinitionEntity创建了流程实例 ExecutionEntity：

ExecutionEntity

ExecutionEntity实现了一些重要接口：

• PVM相关的接口，赋予了ExecutionEntity流程驱动的能力，例如single、start方法。
• 实现VariableScope接口让ExecutionEntity可以持久上下文变量。
• ProcessInstance接口暴露了ExecutionEntity关联的ProcessDefinitionEntity的信息。
• PersistentObject接口代表ExecutionEntity对象是需要持久化。

在ExecutionEntity中维护类一个属性：activity。activity属性代表当前执行到哪个节点，在创建ExecutionEntity过程中会设置activity，使流程从某一个节点开始，默认是开始节点。

最后StartProcessInstanceCmd还调用ExecutionEntity的start方法开始驱动流程：

public void start() {
    performOperation(AtomicOperation.PROCESS_START);
}

驱动流程

Activiti框架的流程运行于PVM模型之上，在流程运行时主要涉及到PVM中几个对象：ActivityImpl、TransitionImpl和ActivityBehavior。

• ActivityImpl：ActivityImpl是流程节点的抽象，ActivityImpl维护流程图中节点的连线，包括有哪些进线，有哪些出线。另外还包含节点同步／异步执行等信息。
• TransitionImpl：TransitionImpl包含source和target两个属性，连接了两个流程节点。
• ActivityBehavior：每一个ActivityImpl对象都拥有一个ActivityBehavior对象，ActivityBehavior代表节点的行为。

ActivityImpl、TransitionImpl和ActivityBehavior只是描述了流程的节点、迁移线和节点行为，真正要让ExecutionEntity流转起来，还需要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();

在ExecutionEntity的start方法中，调用了PROCESS_START，PROCESS_START做了几件事：

• 获取流程定义级别定义的监听start事件的ExecutionListener，调用notify方法。
• 如果开启了事件功能，发布ActivitiEntityWithVariablesEvent和ActivitiProcessStartedEvent。
• 调用PROCESS_START_INITIAL。

PROCESS_START_INITIAL也实现了类似的功能：

• 获取初始节点上定义的监听start事件的ExecutionListener，调用notify方法。
• 调用ACTIVITY_EXECUTE。

在Demo流程执行中涉及的AtomicOperation的链路主要包括：

• ACTIVITY_EXECUTE：调用当前activity的behavior。
• TRANSITION_NOTIFY_LISTENER_END：某个activity节点执行完毕，调用节点上声明的监听end事件的ExecutionListener。
• TRANSITION_NOTIFY_LISTENER_TAKE：触发线上的ExecutionListener。
• TRANSITION_NOTIFY_LISTENER_START：某个activity节点即将开始执行，调用节点上的监听start事件的ExecutionListener。

以Demo流程中的ServiceTask节点helloworld为例，在执行ACTIVITY_EXECUTE时，会获取activity关联的behavior：

public class AtomicOperationActivityExecute implements AtomicOperation {
    public void execute(InterpretableExecution execution) {
        ...
        ActivityImpl activity = (ActivityImpl) execution.getActivity();
        ActivityBehavior activityBehavior = activity.getActivityBehavior();
        activityBehavior.execute(execution);
        ...
    }
}

ServiceTask解析时关联的是ServiceTaskJavaDelegateActivityBehavior，execution方法：

public void execute(ActivityExecution execution) throws Exception {
    //execution中调用了me.likeyao.activiti.demo.HelloWorld
    execute((DelegateExecution) execution);
    //离开当前节点
    leave(execution);
}

在leave方法中调用了：

bpmnActivityBehavior.performDefaultOutgoingBehavior(execution);

performDefaultOutgoingBehavior方法会在当前activity的 出线中选择一条，使流程流向下一个节点。在Demo中只有一条线存在：

protected void performOutgoingBehavior(ActivityExecution execution, 
          boolean checkConditions, boolean throwExceptionIfExecutionStuck, List<ActivityExecution> reusableExecutions) {

    if (transitionsToTake.size() == 1) {
        execution.take(transitionsToTake.get(0));
    }         
          
}

最终take方法会将流程驱动权交还到AtomicOperation中：

public class ExecutionEntity{
    ...
    public void take(PvmTransition transition, boolean fireActivityCompletionEvent) {
        ...
        setActivity((ActivityImpl)transition.getSource());
        setTransition((TransitionImpl) transition);
        performOperation(AtomicOperation.TRANSITION_NOTIFY_LISTENER_END);
        ...
    }
    ...
}

AtomicOperation的问题

按照AtomicOperation的驱动模式，只有当遇到UserTask等需要等待single信号的节点，调用才会返回。这意味着当调用RuntimeService启动一个流程实例时，要一直等到流程运行到一个UserTask节点调用才会返回，如果流程比较长耗时非常验证。

另一个问题是当流程图比较复杂，ExecutionListener数量比较多时，AtomicOperation之间的互相调用会导致调用栈非常深。

AtomicOperation驱动模式与ExecutionEntity、Behavior等绑定的比较紧密，暂时没有特别好的办法替换掉。

小结

本文主要介绍了Activiti框架的启动、部署、运行的主链路，并没有深入BPMN规范和Activit功能的具体实现，后续打算根据Activiti的用户手册，详细分析每个功能的使用和实现。