【开源项目】snakeflow流程引擎研究

开源项目】snakeflow流程引擎研究"/>

【开源项目】snakeflow流程引擎研究

项目地址

（推荐）

常用API

部署流程

processId = engine.process().deploy(StreamHelper.getStreamFromClasspath("test/task/simple/leave.snaker"), null, 10001L);

创建流程实例

Order order = engine.startInstanceById(processId, "1", args);

执行任务

 List<Task> tasks = engine.executeTask(activeTasks.get(0).getId(), "1");

获取某个人的需审批任务

 List<Task> activeTasks2 = engine.query().getActiveTasks(new QueryFilter().setOperator("admin"));

请假流程xml配置

<process displayName="请假流程测试" instanceUrl="/snaker/flow/all" name="leave">
<start displayName="start1" layout="24,124,-1,-1" name="start1" postInterceptors="test.task.interceptor.LocalTaskInterceptor" preInterceptors="test.task.interceptor.LocalTaskInterceptor">
<transition g="" name="transition1" offset="0,0" to="apply"/>
</start>
<end displayName="end1" layout="570,124,-1,-1" name="end1"/>
<task assignee="apply.operator" displayName="请假申请" form="/flow/leave/apply" layout="117,122,-1,-1" name="apply" performType="ANY">
<transition g="" name="transition2" offset="0,0" to="approveDept"/>
</task>
<task assignee="approveDept.operator" displayName="部门经理审批" form="/flow/leave/approveDept" layout="272,122,-1,-1" name="approveDept" performType="ANY">
<transition g="" name="transition3" offset="0,0" to="decision1"/>
</task>
<decision displayName="decision1" expr="#day &gt; 2 ? 'transition5' : 'transition4'" layout="426,124,-1,-1" name="decision1">
<transition displayName="&lt;=2天" g="" name="transition4" offset="0,0" to="end1"/>
<transition displayName="&gt;2天" g="" name="transition5" offset="0,0" to="approveBoss"/>
</decision>
<task assignee="approveBoss.operator" displayName="总经理审批" form="/flow/leave/approveBoss" layout="404,231,-1,-1" name="approveBoss" performType="ANY">
<transition g="" name="transition6" offset="0,0" to="end1"/>
</task>
</process>

流程解析

创建流程实例

SnakerEngineImpl#startInstanceById()，获取StartModel对象，执行execute方法。

	public Order startInstanceById(String id, String operator, Map<String, Object> args) {if(args == null) args = new HashMap<String, Object>();Process process = process().getProcessById(id);process().check(process, id);return startProcess(process, operator, args);}private Order startProcess(Process process, String operator, Map<String, Object> args) {Execution execution = execute(process, operator, args, null, null);if(process.getModel() != null) {StartModel start = process.getModel().getStart();AssertHelper.notNull(start, "流程定义[name=" + process.getName() + ", version=" + process.getVersion() + "]没有开始节点");start.execute(execution);}return execution.getOrder();}

NodeModel#execute，执行拦截器和exec方法。

	public void execute(Execution execution) {intercept(preInterceptorList, execution);exec(execution);intercept(postInterceptorList, execution);}

StartModel#exec

	protected void exec(Execution execution) {runOutTransition(execution);}

NodeModel#runOutTransition，获取TransitionModel对象，执行TransitionModel#execute方法。

	/*** 运行变迁继续执行* @param execution 执行对象*/protected void runOutTransition(Execution execution) {for (TransitionModel tm : getOutputs()) {tm.setEnabled(true);tm.execute(execution);}}

TransitionModel#execute，如果下一个节点是TaskModel对象，执行CreateTaskHandler

	public void execute(Execution execution) {if(!enabled) return;if(target instanceof TaskModel) {//如果目标节点模型为TaskModel，则创建taskfire(new CreateTaskHandler((TaskModel)target), execution);} else if(target instanceof SubProcessModel) {//如果目标节点模型为SubProcessModel，则启动子流程fire(new StartSubProcessHandler((SubProcessModel)target), execution);} else {//如果目标节点模型为其它控制类型，则继续由目标节点执行target.execute(execution);}}

创建任务

CreateTaskHandler#handle

	/*** 根据任务模型、执行对象，创建下一个任务，并添加到execution对象的tasks集合中*/public void handle(Execution execution) {List<Task> tasks = execution.getEngine().task().createTask(model, execution);execution.addTasks(tasks);/*** 从服务上下文中查找任务拦截器列表，依次对task集合进行拦截处理*/List<SnakerInterceptor> interceptors = ServiceContext.getContext().findList(SnakerInterceptor.class);try {for(SnakerInterceptor interceptor : interceptors) {interceptor.intercept(execution);}} catch(Exception e) {log.error("拦截器执行失败=" + e.getMessage());throw new SnakerException(e);}}

TaskService#createTask，获取任务的执行者，根据performType来生成多个任务还是单个任务。

	public List<Task> createTask(TaskModel taskModel, Execution execution) {List<Task> tasks = new ArrayList<Task>();Map<String, Object> args = execution.getArgs();if(args == null) args = new HashMap<String, Object>();Date expireDate = DateHelper.processTime(args, taskModel.getExpireTime());Date remindDate = DateHelper.processTime(args, taskModel.getReminderTime());String form = (String)args.get(taskModel.getForm());String actionUrl = StringHelper.isEmpty(form) ? taskModel.getForm() : form;String[] actors = getTaskActors(taskModel, execution);args.put(Task.KEY_ACTOR, StringHelper.getStringByArray(actors));Task task = createTaskBase(taskModel, execution);task.setActionUrl(actionUrl);task.setExpireDate(expireDate);task.setExpireTime(DateHelper.parseTime(expireDate));task.setVariable(JsonHelper.toJson(args));if(taskModel.isPerformAny()) {//任务执行方式为参与者中任何一个执行即可驱动流程继续流转，该方法只产生一个tasktask = saveTask(task, actors);task.setRemindDate(remindDate);tasks.add(task);} else if(taskModel.isPerformAll()){//任务执行方式为参与者中每个都要执行完才可驱动流程继续流转，该方法根据参与者个数产生对应的task数量for(String actor : actors) {Task singleTask;try {singleTask = (Task) task.clone();} catch (CloneNotSupportedException e) {singleTask = task;}singleTask = saveTask(singleTask, actor);singleTask.setRemindDate(remindDate);tasks.add(singleTask);}}return tasks;}

TaskService#getTaskActors()，获取配置中的assignee，如果参数中有定义args.put("apply.operator", "1");，则是对应key的value值。

	private String[] getTaskActors(TaskModel model, Execution execution) {Object assigneeObject = null;AssignmentHandler handler = model.getAssignmentHandlerObject();if(StringHelper.isNotEmpty(model.getAssignee())) {assigneeObject = execution.getArgs().get(model.getAssignee());} else if(handler != null) {if(handler instanceof Assignment) {assigneeObject = ((Assignment)handler).assign(model, execution);} else {assigneeObject = handler.assign(execution);}}return getTaskActors(assigneeObject == null ? model.getAssignee() : assigneeObject);}

TaskService#getTaskActors(java.lang.Object)，如果拿到的数据是带有英文逗号的字符串，会根据逗号做切割。

	private String[] getTaskActors(Object actors) {if(actors == null) return null;String[] results;if(actors instanceof String) {//如果值为字符串类型，则使用逗号,分隔return ((String)actors).split(",");} else if(actors instanceof List){//jackson会把stirng[]转成arraylist，此处增加arraylist的逻辑判断,by 红豆冰沙2014.11.21List<?> list = (List)actors;results = new String[list.size()];for(int i = 0; i < list.size(); i++) {results[i] = (String)list.get(i);}return results;} else if(actors instanceof Long) {//如果为Long类型，则返回1个元素的String[]results = new String[1];results[0] = String.valueOf((Long)actors);return results;} else if(actors instanceof Integer) {//如果为Integer类型，则返回1个元素的String[]results = new String[1];results[0] = String.valueOf((Integer)actors);return results;} else if(actors instanceof String[]) {//如果为String[]类型，则直接返回return (String[])actors;} else {//其它类型，抛出不支持的类型异常throw new SnakerException("任务参与者对象[" + actors + "]类型不支持."+ "合法参数示例:Long,Integer,new String[]{},'10000,20000',List<String>");}}

判断模型

DecisionModel#exec，获取模型中的expr属性，执行判断。根据执行结果决定执行哪一个TransitionModel

	public void exec(Execution execution) {log.info(execution.getOrder().getId() + "->decision execution.getArgs():" + execution.getArgs());if(expression == null) {expression = ServiceContext.getContext().find(Expression.class);}log.info("expression is " + expression);if(expression == null) throw new SnakerException("表达式解析器为空，请检查配置.");String next = null;if(StringHelper.isNotEmpty(expr)) {next = expression.eval(String.class, expr, execution.getArgs());} else if(decide != null) {next = decide.decide(execution);}log.info(execution.getOrder().getId() + "->decision expression[expr=" + expr + "] return result:" + next);boolean isfound = false;for(TransitionModel tm : getOutputs()) {if(StringHelper.isEmpty(next)) {String expr = tm.getExpr();if(StringHelper.isNotEmpty(expr) && expression.eval(Boolean.class, expr, execution.getArgs())) {tm.setEnabled(true);tm.execute(execution);isfound = true;}} else {if(tm.getName().equals(next)) {tm.setEnabled(true);tm.execute(execution);isfound = true;}}}if(!isfound) throw new SnakerException(execution.getOrder().getId() + "->decision节点无法确定下一步执行路线");}

解析xml成ProcessModel对象

ProcessService#deploy()，根据ModelParser进行解析，使用缓存避免二次解析造成时间浪费。

    public String deploy(InputStream input, String creator, Long tenantId) {AssertHelper.notNull(input);try {byte[] bytes = StreamHelper.readBytes(input);ProcessModel model = ModelParser.parse(bytes);Integer version = access().getLatestProcessVersion(model.getName());Process entity = new Process();entity.setId(StringHelper.getPrimaryKey());if (version == null || version < 0) {entity.setVersion(0);} else {entity.setVersion(version + 1);}entity.setState(STATE_ACTIVE);entity.setModel(model);entity.setBytes(bytes);entity.setCreateTime(DateHelper.getTime());entity.setCreator(creator);entity.setTenantId(tenantId);saveProcess(entity);cache(entity);return entity.getId();} catch (Exception e) {e.printStackTrace();log.error(e.getMessage());throw new SnakerException(e.getMessage(), e.getCause());}}

ModelParser#parse，获取xml中的节点数据

	public static ProcessModel parse(byte[] bytes) {DocumentBuilder documentBuilder = XmlHelper.createDocumentBuilder();if(documentBuilder != null) {Document doc = null;try {doc = documentBuilder.parse(new ByteArrayInputStream(bytes));Element processE = doc.getDocumentElement();ProcessModel process = new ProcessModel();process.setName(processE.getAttribute(NodeParser.ATTR_NAME));process.setDisplayName(processE.getAttribute(NodeParser.ATTR_DISPLAYNAME));process.setExpireTime(processE.getAttribute(NodeParser.ATTR_EXPIRETIME));process.setInstanceUrl(processE.getAttribute(NodeParser.ATTR_INSTANCEURL));process.setInstanceNoClass(processE.getAttribute(NodeParser.ATTR_INSTANCENOCLASS));NodeList nodeList = processE.getChildNodes();int nodeSize = nodeList.getLength();for(int i = 0; i < nodeSize; i++) {Node node = nodeList.item(i);if (node.getNodeType() == Node.ELEMENT_NODE) {NodeModel model = parseModel(node);process.getNodes().add(model);}}//循环节点模型，构造变迁输入、输出的source、targetfor(NodeModel node : process.getNodes()) {for(TransitionModel transition : node.getOutputs()) {String to = transition.getTo();for(NodeModel node2 : process.getNodes()) {if(to.equalsIgnoreCase(node2.getName())) {node2.getInputs().add(transition);transition.setTarget(node2);}}}}return process;} catch (SAXException e) {e.printStackTrace();throw new SnakerException(e);} catch (IOException e) {throw new SnakerException(e);}} else {throw new SnakerException("documentBuilder is null");}}

补充

数据操作

DBAccess数据库操作接口类，用于保存任务，流程和流程实例。AbstractDBAccess实现了DBAccess，定义了修改和插入数据的sql。如果想要做租户化定制，可以在AbstractDBAccess做修改。

/*** 数据库访问接口* 主要提供保存、更新、查询流程的相关table* @author yuqs* @since 1.0*/
public interface DBAccess {/*** 根据访问对象，设置具体的实现类* @param accessObject 数据库访问对象(Connection等)*/public void initialize(Object accessObject);/*** 保存任务对象* @param task 任务对象*/public void saveTask(Task task);/*** 保存流程实例对象* @param order 流程实例对象*/public void saveOrder(Order order);/*** 保存抄送实例* @param ccorder 抄送实体* @since 1.5*/public void saveCCOrder(CCOrder ccorder);/*** 保存流程定义对象* @param process 流程定义对象*/public void saveProcess(Process process);/*** 保存任务参与者对象* @param taskActor 任务参与者对象*/public void saveTaskActor(TaskActor taskActor);/*** 更新任务对象* @param task 任务对象*/public void updateTask(Task task);/*** 更新流程实例对象* @param order 流程实例对象*/public void updateOrder(Order order);/*** 更新抄送状态* @param ccorder 抄送实体对象*/public void updateCCOrder(CCOrder ccorder);/*** 更新流程定义对象* @param process 流程定义对象*/public void updateProcess(Process process);/*** 删除流程定义对象* @param process 流程定义对象*/public void deleteProcess(Process process);/*** 更新流程定义类别* @param type 类别* @since 1.5*/public void updateProcessType(String id, String type);/*** 删除任务、任务参与者对象* @param task 任务对象*/public void deleteTask(Task task);/*** 删除流程实例对象* @param order 流程实例对象*/public void deleteOrder(Order order);/*** 删除抄送记录* @param ccorder 抄送实体对象*/public void deleteCCOrder(CCOrder ccorder);/*** 删除参与者* @param taskId 任务id* @param actors 参与者集合*/public void removeTaskActor(String taskId, String... actors);/*** 迁移活动实例* @param order 历史流程实例对象*/public void saveHistory(HistoryOrder order);/*** 更新历史流程实例状态* @param order 历史流程实例对象*/public void updateHistory(HistoryOrder order);/*** 迁移活动任务* @param task 历史任务对象*/public void saveHistory(HistoryTask task);/*** 删除历史实例记录* @param historyOrder 历史实例*/public void deleteHistoryOrder(HistoryOrder historyOrder);/*** 删除历史任务记录* @param historyTask 历史任务*/public void deleteHistoryTask(HistoryTask historyTask);/*** 更新实例变量（包括历史实例表）* @param order 实例对象*/public void updateOrderVariable(Order order);/*** 保存委托代理对象* @param surrogate 委托代理对象*/public void saveSurrogate(Surrogate surrogate);/*** 更新委托代理对象* @param surrogate 委托代理对象*/public void updateSurrogate(Surrogate surrogate);/*** 删除委托代理对象* @param surrogate 委托代理对象*/public void deleteSurrogate(Surrogate surrogate);/*** 根据主键id查询委托代理对象* @param id 主键id* @return surrogate 委托代理对象*/public Surrogate getSurrogate(String id);/*** 根据授权人、流程名称查询委托代理对象* @param page 分页对象* @param filter 查询过滤器* @return List<Surrogate> 委托代理对象集合*/public List<Surrogate> getSurrogate(Page<Surrogate> page, QueryFilter filter);/*** 根据任务id查询任务对象* @param taskId 任务id* @return Task 任务对象*/public Task getTask(String taskId);/*** 根据任务ID获取历史任务对象* @param taskId 历史任务id* @return 历史任务对象*/HistoryTask getHistTask(String taskId);/*** 根据父任务id查询所有子任务* @param parentTaskId 父任务id* @return List<Task> 活动任务集合*/public List<Task> getNextActiveTasks(String parentTaskId);/*** 根据流程实例id、任务名称获取* @param orderId 流程实例id* @param taskName 任务名称* @param parentTaskId 父任务id* @return List<Task> 活动任务集合*/public List<Task> getNextActiveTasks(String orderId, String taskName, String parentTaskId);/*** 根据任务id查询所有活动任务参与者集合* @param taskId 活动任务id* @return List<TaskActor> 活动任务参与者集合*/public List<TaskActor> getTaskActorsByTaskId(String taskId);/*** 根据任务id查询所有历史任务参与者集合* @param taskId 历史任务id* @return List<HistoryTaskActor> 历史任务参与者集合*/public List<HistoryTaskActor> getHistTaskActorsByTaskId(String taskId);/*** 根据流程实例id查询实例对象* @param orderId 活动流程实例id* @return Order 活动流程实例对象*/public Order getOrder(String orderId);/*** 根据流程实例id、参与者id获取抄送记录* @param orderId 活动流程实例id* @param actorIds 参与者id* @return 传送记录列表*/public List<CCOrder> getCCOrder(String orderId, String... actorIds);/*** 根据流程实例ID获取历史流程实例对象* @param orderId 历史流程实例id* @return HistoryOrder 历史流程实例对象*/HistoryOrder getHistOrder(String orderId);/*** 根据流程定义id查询流程定义对象* @param id 流程定义id* @return Process 流程定义对象*/public Process getProcess(String id);/*** 根据流程名称查询最近的版本号* @param name 流程名称* @return Integer 流程定义版本号*/public Integer getLatestProcessVersion(String name);/*** 根据查询的参数，分页对象，返回分页后的查询结果* @param page 分页对象* @param filter 查询过滤器* @return List<Process> 流程定义集合*/public List<Process> getProcesss(Page<Process> page, QueryFilter filter);/*** 分页查询流程实例* @param page 分页对象* @param filter 查询过滤器* @return List<Order> 活动流程实例集合*/public List<Order> getActiveOrders(Page<Order> page, QueryFilter filter);/*** 分页查询活动任务列表* @param page 分页对象* @param filter 查询过滤器* @return List<Task> 活动任务集合*/public List<Task> getActiveTasks(Page<Task> page, QueryFilter filter);/*** 分页查询历史流程实例* @param page 分页对象* @param filter 查询过滤器* @return List<HistoryOrder> 历史流程实例集合*/public List<HistoryOrder> getHistoryOrders(Page<HistoryOrder> page, QueryFilter filter);/*** 根据参与者分页查询已完成的历史任务* @param page 分页对象* @param filter 查询过滤器* @return List<HistoryTask> 历史任务集合*/public List<HistoryTask> getHistoryTasks(Page<HistoryTask> page, QueryFilter filter);/*** 根据查询的参数，分页对象，返回分页后的活动工作项* @param page 分页对象* @param filter 查询过滤器* @return List<WorkItem> 活动工作项*/public List<WorkItem> getWorkItems(Page<WorkItem> page, QueryFilter filter);/*** 根据查询的参数，分页对象，返回分页后的抄送任务项* @param page 分页对象* @param filter 查询过滤器* @return List<WorkItem> 活动工作项*/public List<HistoryOrder> getCCWorks(Page<HistoryOrder> page, QueryFilter filter);/*** 根据流程定义ID、参与者分页查询已完成的历史任务项* @param page 分页对象* @param filter 查询过滤器* @return List<WorkItem> 历史工作项*/public List<WorkItem> getHistoryWorkItems(Page<WorkItem> page, QueryFilter filter);/*** 根据类型clazz、Sql语句、参数查询单个对象* @param clazz 类型* @param sql sql语句* @param args 参数列表* @return 结果对象*/public <T> T queryObject(Class<T> clazz, String sql, Object... args);/*** 根据类型clazz、Sql语句、参数查询列表对象* @param clazz 类型* @param sql sql语句* @param args 参数列表* @return 结果对象列表*/public <T> List<T> queryList(Class<T> clazz, String sql, Object... args);/*** 根据类型clazz、Sql语句、参数分页查询列表对象* @param page 分页对象* @param filter 查询过滤器* @param clazz 类型* @param sql sql语句* @param args 参数列表* @return 结果对象列表*/public <T> List<T> queryList(Page<T> page, QueryFilter filter, Class<T> clazz, String sql, Object... args);/*** 运行脚本文件*/public void runScript();
}

Context

public interface Context {/*** 根据服务名称、实例向服务工厂注册* @param name 服务名称* @param object 服务实例*/void put(String name, Object object);/*** 根据服务名称、类型向服务工厂注册* @param name 服务名称* @param clazz 类型*/void put(String name, Class<?> clazz);/*** 判断是否存在给定的服务名称* @param name 服务名称* @return*/boolean exist(String name);/*** 根据给定的类型查找服务实例* @param clazz 类型* @return*/<T> T find(Class<T> clazz);/*** 根据给定的类型查找所有此类型的服务实例* @param clazz 类型* @return*/<T> List<T> findList(Class<T> clazz);/*** 根据给定的服务名称、类型查找服务实例* @param name 服务名称* @param clazz 类型* @return*/<T> T findByName(String name, Class<T> clazz);
}

服务容器类

常用的形式是DBAccess access = ServiceContext.find(DBAccess.class);，

public interface Context {/*** 根据服务名称、实例向服务工厂注册* @param name 服务名称* @param object 服务实例*/void put(String name, Object object);/*** 根据服务名称、类型向服务工厂注册* @param name 服务名称* @param clazz 类型*/void put(String name, Class<?> clazz);/*** 判断是否存在给定的服务名称* @param name 服务名称* @return*/boolean exist(String name);/*** 根据给定的类型查找服务实例* @param clazz 类型* @return*/<T> T find(Class<T> clazz);/*** 根据给定的类型查找所有此类型的服务实例* @param clazz 类型* @return*/<T> List<T> findList(Class<T> clazz);/*** 根据给定的服务名称、类型查找服务实例* @param name 服务名称* @param clazz 类型* @return*/<T> T findByName(String name, Class<T> clazz);
}

Configuration#parser()，解析snaker.xml和base.config.xml

	protected void parser() {if(log.isDebugEnabled()) {log.debug("Service parsing start......");}//默认使用snaker.xml配置自定义的beanString config = ConfigHelper.getProperty("config");if (StringHelper.isEmpty(config)) {config = USER_CONFIG_FILE;}parser(config);parser(BASE_CONFIG_FILE);if (!isCMB()) {parser(EXT_CONFIG_FILE);for(Entry<String, Class<?>> entry : txClass.entrySet()) {if(interceptor != null) {Object instance = interceptor.getProxy(entry.getValue());ServiceContext.put(entry.getKey(), instance);} else {ServiceContext.put(entry.getKey(), entry.getValue());}}}if(log.isDebugEnabled()) {log.debug("Service parsing finish......");}}

Configuration#parser(java.lang.String)，解析xml文件，加载xml配置的class对象到容器中。

	private void parser(String resource) {//解析所有配置节点，并实例化class指定的类DocumentBuilder documentBuilder = XmlHelper.createDocumentBuilder();try {if (documentBuilder != null) {InputStream input = StreamHelper.openStream(resource);if(input == null) return;Document doc = documentBuilder.parse(input);Element configElement = doc.getDocumentElement();NodeList nodeList = configElement.getChildNodes();int nodeSize = nodeList.getLength();for(int i = 0; i < nodeSize; i++) {Node node = nodeList.item(i);if (node.getNodeType() == Node.ELEMENT_NODE) {Element element = (Element)node;String name = element.getAttribute("name");String className = element.getAttribute("class");String proxy = element.getAttribute("proxy");if(StringHelper.isEmpty(name)) {name = className;}if(ServiceContext.exist(name)) {log.warn("Duplicate name is:" + name);continue;}Class<?> clazz = ClassHelper.loadClass(className);if(TransactionInterceptor.class.isAssignableFrom(clazz)) {interceptor = (TransactionInterceptor)ClassHelper.instantiate(clazz);ServiceContext.put(name, interceptor);continue;}if(proxy != null && proxy.equalsIgnoreCase("transaction")) {txClass.put(name, clazz);} else {ServiceContext.put(name, clazz);}}}}} catch (Exception e) {e.printStackTrace();throw new SnakerException("资源解析失败，请检查配置文件[" + resource + "]", e.getCause());}}