简单流程执行

上文中已经对模型进行了高度的抽象，本文将对模型行为进行进一步说明。这里以提问的方式开场：

如何让流程从开始节点按箭头指向走到结束节点？

StartModel(start)->TaskModel(apply)->TaskModel(deptApprove)->EndModel(end)

执行过程分析

对象图：

时序图：

执行过程说明：

1. 开始节点调用输出边t1的执行方法
2. t1输出边调用请假申请节点的执行方法
3. 请假节点调用输出边t2的执行方法
4. t2输出边调用部门领导审批的执行方法
5. 部门领导审批调用输出边t3的执行方法
6. t3调用结束节点的执行方法

代码实现步骤

节点模型的execute方法增加打印当前对象的编码和名称并增加构造函数

class NodeModel (ABC, BaseModel, Action):
    """
    节点模型
    """
    __abstract__ = True
    # 输入边集合
    inputs = []
    # 输出边集合
    outputs = []
    # 前置拦截器-字符串，多个使用英文逗号分割
    preInterceptors = None
    # 后置拦截器-字符串，多个使用英文逗号分割
    postInterceptors = None
    def __init__(self):
        """
        构造函数-重新初始化属性
        因python的父类属性会共享，所以需要重新初始化属性，要不然所有的节点属性都指向同一个对象
        """
        self.inputs = []
        self.outputs = []
        self.preInterceptors = None
        self.postInterceptors = None
    @abstractmethod
    def exec(self, execution):
        """
        由子类自定义执行方法
        @param execution: 执行对象参数
        """
    def execute(self, execution):
        """
        执行节点
        @param execution: 执行对象参数
        """
        logging.info(f"model:{self.__class__.__name__},name:{self.name},displayName:{self.displayName}")
        self.exec(execution)

流程模型增加获取开始节点方法

class ProcessModel(BaseModel):
    type = None # 流程定义分类
    instanceUrl = None # 启动实例要填写的表单key
    expireTime = None # 期待完成时间变量key
    instanceNoClass = None # 实例编号生成器实现类
    # 流程定义的所有节点
    nodes = []
    # 流程定义的所有任务节点
    tasks = []
    def get_start(self):
        """
        获取流程模型的开始节点
        """ 
        startModel = None
        for node in self.nodes:
            if isinstance(node, StartModel):
                startModel = node
                break
        return startModel

流程模型拿到开始节点对象并调用执行方法

processModel.get_start().execute({})

此时只打印开始节点信息

model:StartModel,name:start,displayName:开始

节点模型对象的execute方法增加遍历调用下一个节点执行方法的

public abstract class NodeModel extends BaseModel implements Action {
    private String layout;// 布局属性(x,y,w,h)
    // 输入边集合
    private List<TransitionModel> inputs = new ArrayList<TransitionModel>();
    // 输出边集合
    private List<TransitionModel> outputs = new ArrayList<TransitionModel>();
    private String preInterceptors; // 节点前置拦截器
    private String postInterceptors; // 节点后置拦截器
    
    /**
    * 由子类自定义执行方法
    * @param execution
    */
    abstract void exec(Execution execution);
    @Override
    public void execute(Execution execution) {
        // 1. 调用前置拦截器
        // 2. 调用子类的exec方法
        // 3. 调用后置拦截器
        System.out.println(StrUtil.format("model:{},name:{},displayName:{}", this.getClass().getSimpleName(), getName(),getDisplayName()));
        outputs.forEach(tr->{
            tr.getTarget().execute(execution);
        });
        exec(execution);
    }
}

结果：

model:StartModel,name:start,displayName:开始
model:TaskModel,name:apply,displayName:请假申请
model:TaskModel,name:deptApprove,displayName:部门领导审批
model:EndModel,name:end,displayName:结束

为了突显边的作用，我们可以实现边的执行方法：

class TransitionModel(BaseModel,Action):
    source = None # 边源节点引用
    target = None # 边目标节点引用
    to = None # 目标节点名称
    expr = None # 边表达式
    enabled = False # 是否可执行
    def execute(self, execution):
        if not self.enabled: return
        self.target.execute(execution)

然后改造节点模型增加run_out_transition方法

class NodeModel (ABC, BaseModel, Action):
    """
    节点模型
    """
    __abstract__ = True
    # 输入边集合
    inputs = []
    # 输出边集合
    outputs = []
    # 前置拦截器-字符串，多个使用英文逗号分割
    preInterceptors = None
    # 后置拦截器-字符串，多个使用英文逗号分割
    postInterceptors = None
    def __init__(self):
        """
        构造函数-重新初始化属性
        因python的父类属性会共享，所以需要重新初始化属性，要不然所有的节点属性都指向同一个对象
        """
        self.inputs = []
        self.outputs = []
        self.preInterceptors = None
        self.postInterceptors = None

    @abstractmethod
    def exec(self, execution):
        """
        由子类自定义执行方法
        @param execution: 执行对象参数
        """
    def execute(self, execution):
        """
        执行节点
        @param execution: 执行对象参数
        """
        logging.info(f"model:{self.__class__.__name__},name:{self.name},displayName:{self.displayName}")
        self.run_out_transition(execution)
        self.exec(execution)
    def run_out_transition(self, execution):
        """
        执行输出边
        """
        for transition in self.outputs:
            transition.enabled = True
            transition.execute(execution)

最终效果为：

model:StartModel,name:start,displayName:开始
model:TaskModel,name:apply,displayName:请假申请
model:TaskModel,name:deptApprove,displayName:部门领导审批
model:EndModel,name:end,displayName:结束

如何让流程产生阻塞？

上面的例子执行过程太顺利了，真实的工作流场景会存在一些阻塞任务，产生阻塞的意思是，即调用节点执行方法，如果条件不满足，依然不能驱动流程往下一个节点进行。那我们如何使用程序去模拟这一过程呢？

首先改造节点模型

并不是每个节点的执行方式都一样，我们需要对不同节点进行不同的输出处理，所以这里

• 暂时去掉原来节点模型的打印语句和调用执行边的方法
• repr()方法

from abc import ABC, abstractmethod
from datetime import datetime
class BaseModel (object):
    """
    基础模型
    """
    __abstract__ = True
    # 唯一编码
    name = None
    # 显示名称
    displayName = None
class Action (object):
    """
    节点行为接口
    """
    __abstract__ = True
    def execute(self, execution):
        """
        执行行为
        @param execution: 执行对象参数
        """
class NodeModel (ABC, BaseModel, Action):
    """
    节点模型
    """
    __abstract__ = True
    # 输入边集合
    inputs = []
    # 输出边集合
    outputs = []
    # 前置拦截器-字符串，多个使用英文逗号分割
    preInterceptors = None
    # 后置拦截器-字符串，多个使用英文逗号分割
    postInterceptors = None
    def __init__(self):
        """
        构造函数-重新初始化属性
        因python的父类属性会共享，所以需要重新初始化属性，要不然所有的节点属性都指向同一个对象
        """
        self.inputs = []
        self.outputs = []
        self.preInterceptors = None
        self.postInterceptors = None

    @abstractmethod
    def exec(self, execution):
        """
        由子类自定义执行方法
        @param execution: 执行对象参数
        """
    def execute(self, execution):
        """
        执行节点
        @param execution: 执行对象参数
        """
        self.exec(execution)
    def run_out_transition(self, execution):
        """
        执行输出边
        """
        for transition in self.outputs:
            transition.enabled = True
            transition.execute(execution)
    def __repr__(self) -> str:
        """
        重写输出
        """
        return f"model:{self.__class__.__name__},name:{self.name},displayName:{self.displayName},time:{datetime.now()}"

实现开始节点的exec方法

开始节点的exec主要执行如下逻辑：

• 输出self
• 调用run_out_transition

public class StartModel extends NodeModel {
    @Override
    void exec(Execution execution) {
        System.out.println(self);
        runOutTransition(execution);
    }
}

实现结束节点的exec方法

结束节点是没有输出边的，所以只输出self

class EndModel(NodeModel):
    """
    结束节点模型
    """
    def exec(self, execution):
        """
        执行节点
        @param execution: 执行对象参数
        """
        # 执行结束节点自定义执行逻辑
        logging.info(self)

实现任务节点的exec方法

任务节点的比较特殊，我们可以做如下处理让其产生临时的阻塞：

class TaskModel(NodeModel):
    """
    任务节点模型
    """
    form = None # 表单标识
    assignee = None # 参与人
    assignmentHandler = None # 参与人处理类
    taskType = None # 任务类型(主办/协办)
    performType = None # 参与类型(普通参与/会签参与)
    reminderTime = None # 提醒时间
    reminderRepeat = None # 重复提醒间隔
    expireTime = None # 期待任务完成时间变量key
    autoExecute = None # 到期是否自动执行Y/N
    callback = None # 自动执行回调类
    ext = {} # 自定义扩展属性
    def exec(self, execution):
        """
        执行节点
        @param execution: 执行对象参数
        """
        # 执行任务节点自定义执行逻辑
        time.sleep(3)
        logging.info(self)
        self.run_out_transition(execution)

此时打印结果如下：

model:StartModel,name:start,displayName:开始,time:2024-07-22 20:43:25.379367
model:TaskModel,name:apply,displayName:请假申请,time:2024-07-22 20:43:28.389158
model:TaskModel,name:deptApprove,displayName:部门领导审批,time:2024-07-22 20:43:31.390409
model:EndModel,name:end,displayName:结束,time:2024-07-22 20:43:31.390409

相关源码

工作流引擎核心设计·简单流程执行