工作流引擎原理-打造一款适合自己的工作流引擎

发表于

前言

作为开发人员或多或少都会接触过工作流引擎，如 activiti、 jbpm等，这些工作流引擎都会比较重。在小项目中引入就会显得不是很优雅。本文主要简单介绍工作流引擎的原理并讲述如何慢慢打造一款适合自己的工作流引擎。

流程节点

当工程需若干个不同程序(流程)或分若干个阶段来完成时，某一程序或某一阶段结束，另一程序或某一阶段开始时的转接点(类别点或时间点)，称流程节点。 常见的流程节点有：开始结束、任务节点、条件节点、分支节点、合并节点、子流程节点、结束节点等。为了简单起见，这里只讲解最简单的流程（开始->任务->结束）。

开始节点

流程的开始，做一些初始化操作的节点。

  • 类型 START
  • 模型类 StartModel
  • 属性说明
属性名 类型 说明
id String 节点id，流程节点的唯一标识
name String 节点名称，说明是节点做什么的
nextNodeId String 下一个节点id
ext Map 扩展属性

任务节点

该节点会产生任务，任务完成后才能进行下一步

  • 类型 TASK
  • 模型类 TaskModel
  • 属性说明
属性名 类型 说明
id String 节点id，流程节点的唯一标识
name String 节点名称，说明是节点做什么的
nextNodeId String 下一个节点id
performType String 参与方式（ANY->任何一个参与者处理完即可执行下一步,ALL->所有参与者都完成，才可执行下一步）
expireTime String 期望完成时间（产生的任务期望啥时间完成）
reminderTime String 提醒时间（如任务未处理，啥时候提醒）
reminderRepeat Integer 提醒间隔(分钟)（如任务未处理，提醒规则是什么）
autoExecute Boolean 是否自动执行（如任务未处理且到期，是否自动执行）
ext Map 扩展属性

结束节点

该节点说明流程已走完，在这里触发流程完成事件

  • 类型 END
  • 模型类 EndModel
  • 属性说明
属性名 类型 说明
id String 节点id，流程节点的唯一标识
name String 节点名称，说明是节点做什么的
ext Map 扩展属性

流程定义样例

  • 新闻中心稿件发布流程
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
复制代码{
	"flowId":"cmsNews",
	"flowName":"新闻中心稿件发布",
	"nodeList": [
		{
			"id":"start",
			"nodeType": "START",
			"name":"开始",
			"nextNodeId": "collect"
		},
		{
			"id":"collect",
			"nodeType": "TASK",
			"name":"采编新闻稿件",
			"performType":"ANY",
			"nextNodeId": "firstReview",
			"ext":{
				"userIds":"1,2,3"
			}
		},
		{
			"id":"firstReview",
			"nodeType": "TASK",
			"name":"党群部宣传科初审",
			"performType":"ANY",
			"nextNodeId": "secondReview",
			"ext":{
				"userIds":"1,2,3"
			}
		},
		{
			"id":"secondReview",
			"nodeType": "TASK",
			"name":"党群部二审",
			"performType":"ANY",
			"nextNodeId": "end",
			"ext":{
				"userIds":"1,2,3"
			}
		},
		{
			"id":"end",
			"nodeType": "END",
			"name":"结束"
		}
	]
}

表结构设计

  • flow_define(流程定义)
字段 类型 空 默认 注释
id char(36) False 主键
flow_define_code varchar(16) True 流程定义编码
flow_name varchar(100) True 流程名称
description varchar(255) True 流程描述
flow_define mediumtext True 流程定义（json字符串）
create_time datetime True 创建时间
update_time datetime True 更新时间
is_deleted tinyint(1) unsigned True 0 是否删除(1->删除YES,0->未删除NO)
  • flow_work(流程实例)
字段 类型 空 默认 注释
id char(36) False 主键
flow_name varchar(100) True 流程名称
flow_define_code varchar(32) True 流程定义编码
last_operator char(36) True 最后一个操作者id
current_node_id char(36) True 当前节点id
next_node_id char(36) True 下一步节点id
flow_define mediumtext True 流程定义（json字符串）
status int(6) True 流程状态(10->开始START,30->结束END,40->取消CANCEL)
flow_param mediumtext True 流程参数json
create_time datetime True 创建时间
update_time datetime True 更新时间
is_deleted tinyint(1) unsigned True 0 是否删除(1->删除YES,0->未删除NO)
  • flow_task(流程任务)
字段 类型 空 默认 注释
id char(36) False 主键
flow_work_id char(36) True 流程id
flow_node_id char(36) True 流程节点id
task_name varchar(100) True 任务名称
operator char(36) True 操作者用户id
actor_user_id char(36) True 执行用户的id
status varchar(6) True 10 任务状态(10->新建CREATED,20->已完成FINISHED,30->过期EXPIRED,40->取消CANCEL,50->驳回REJECT)
service_type varchar(32) True 业务类型
service_id varchar(36) True 关联的业务id
finish_time datetime True 完成时间
flow_param mediumtext True 流程参数json
create_time datetime True 创建时间
update_time datetime True 更新时间
is_deleted tinyint(1) unsigned True 0 是否删除(1->删除YES,0->未删除NO)

处理流程关键字描述

  • 关于流程定义文件

流程定义文件即描述流程各节点关系的文件，在这里使用json文件来描述，存储在flow_define表的flow_define字段上。

  • 关于流程定义文件解析

流程定义文件解析即将流程定义文件中的各个流程节点解析出来，并将其组装成流程模型。

  • 关于流程实例

流程实例即使用流程定义生成的一个流程实例，生成的流程实例会存储在flow_work表。（该操作是由业务系统发起，即发起一个流程。）

  • 关于任务

任务会在任务节点上产生，即当执行到任务节点时，会使用该节点的处理器，给参与的人员分派任务，分派的任务存储在flow_task表中。业务系统中可通过查询该表，拿到登录用户的待办任务。

  • 关于执行完成任务

当参与人员把待办任务处理后，业务系统会调用完成任务方法，流程引擎会修改任务完成状态，并驱动流程往下一个节点行进。

  • 关于驳回任务

当参与人员驳回任务后，业务系统会调用任务驳回方法，流程引擎会将流程进度调回上一个节点。

核心处理流程描述

FlowWorkCtl.java

  • 启动流程实例
  1. 加载流程定义
  2. 解析流程定义文件
  3. 节点处理器控制器调用执行节点处理
    1. 如果是开始节点，只需要发布开始事件
    2. 如果是任务节点，则根据节点配置，分派任务，发布开始事件并更新流程实例节点状态(非必须，只是简单记录)
    3. 如果是结束节点，则修改流程实例为完成，并发布结束事件。
  • 执行任务
  1. 获取任务实例
  2. 获取流程实例
  3. 解析流程定义文件
  4. 将任务设置为已完成
  5. 判断节点任务是否完成
    1. 如果已完成，则拿到下一个节点，调用节点处理控制器执行
    2. 如未完成，则无需处理
  • 驳回任务
  1. 获取任务实例
  2. 获取流程实例
  3. 解析流程定义文件
  4. 将任务设置为已驳回
  5. 修改节点的其他参与人员的任务为已取消
  6. 拿到上一个节点模型，调用节点处理控制器执行

开始编码

工程目录

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
复制代码com.mole.modules.flow
├── entity	# 实体类，与表结构对应
	├── FlowDefine.java	   # 流程定义实体
	├── FlowTask.java	   # 流程任务实体
	└── FlowWork.java      # 流程实例实体
├── enums        # 错误码枚举
	└── FlowErrEnum.java
├── event          # 事件
	├── FlowEndNodeEvent.java	   # 流程结束事件
	├── FlowStartNodeEvent.java	   # 流程开始事件
	├── FlowTaskCreatedEvent.java  # 流程任务创建事件
	└── FlowTaskNode.Event.java    # 任务开始执行事件
├── mapper	# 持久层
	├── FlowDefineMapper.java
	├── FlowTaskMapper.java
	└── FlowWorkMapper.java
├── param	# 参数实体
	└── FlowParam.java      # 流程参数实体
├── model	# 模型层
	├── BaseModel.java		# 基础节点模型
	├── EndModel.java		# 结束节点模型
	├── FlowModel.java		# 流程模型
	├── StartModel.java		# 开始节点模型
	└── TaskModel.java		# 任务节点模型
├── parser
	└── FlowNodeParser.java # 节点解析器	
├── processor
	├── impl
		├── EndNodeProcessorImpl.java	# 结束节点处理实现类
		├── StartNodeProcessorImpl.java	# 开始节点处理实现类
		└── TaskNodeProcessorImpl.java	# 任务节点处理实现类
	├── FlowNodeProcessor.java		# 节点处理接口
	└── FlowNodeProcessorCtl.java	# 节点处理器控制器
├── service
	└── FlowNodeUserService.java	# 自定义节点参与人接口
└── FlowWorkCtl.java        # 流程控制类-引擎

核心类

模型类

  • FlowModel.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
复制代码package com.mole.modules.flow.model;

import java.io.Serializable;
import java.util.List;
/**
 * 流程模型
 * @author MLD
 *
 */
public class FlowModel implements Serializable{

	/**
	 * 
	 */
	private static final long serialVersionUID = -3978388377128544781L;
	/**
	 * 流程id
	 */
	private String flowId;
	/**
	 * 流程名称
	 */
	private String flowName;
	/**
	 * 流程节点
	 */
	private List<BaseModel> nodeList;
	// ....get .....set 略
	/**
	 * 通过id获取节点
	 * @param nodeId
	 * @return
	 */
	public BaseModel getNodeModel(String nodeId) {
		if(null == nodeList || nodeList.isEmpty()) {
			return null;
		}
		for(BaseModel node : nodeList) {
			if(node.getId().equals(nodeId)){
				return node;
			}
		}
		return null;
	}
	/**
	 * 获取开始节点模型
	 * @return
	 */
	public StartModel getStartModel() {
		if(null == nodeList || nodeList.isEmpty()) {
			return null;
		}
		for(BaseModel node : nodeList) {
			if(node instanceof StartModel){
				return (StartModel)node;
			}
		}
		return null;
	}
    /**
	 * 通过当前节点的上一个节点
	 * @param nodeId
	 * @return
	 */
	public BaseModel getPreNodeModel(String nodeId) {
		if(null == nodeList || nodeList.isEmpty()) {
			return null;
		}
		for(BaseModel node : nodeList) {
			if(nodeId.equals(node.getNextNodeId())){
				return node;
			}
		}
		return null;
	}
	/**
	 * 获取结束节点模型
	 * @return
	 */
	public EndModel getEndModel() {
		if(null == nodeList || nodeList.isEmpty()) {
			return null;
		}
		for(BaseModel node : nodeList) {
			if(node instanceof EndModel){
				return (EndModel)node;
			}
		}
		return null;
	}
	
}
  • BaseModel.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
复制代码package com.mole.modules.flow.model;
import java.io.Serializable;
import java.util.Map
public class BaseModel implements Serializable {
    /**
	 * 节点id
	 */
    private String id;
    /**
	 * 节点名称
	 */
    private String name;
    /**
	 * 节点类型
	 */
	private String nodeType;
    /**
	 * 下一个节点id
	 */
    private String nextNodeId;
    /**
	 * 扩展属性
	 */
    private Map<String,Object> ext;
}
  • StartModel.java
1
2
3
4
5
复制代码package com.mole.modules.flow.model;
import java.io.Serializable;
public class StartModel extends BaseModel implements Serializable {
    
}
  • TaskModel.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
复制代码package com.mole.modules.flow.model;
import java.io.Serializable;
public class TaskModel extends BaseModel implements Serializable {
    /**
	 * 参与方式（
	 * ANY->任何一个参与者处理完即可执行下一步,
	 * ALL->所有参与者都完成，才可执行下一步
	 * ）
	 */
    private String performType;
    /**
	 * 期望完成时间（产生的任务期望啥时间完成）
	 */
    private String expireTime;
    /**
	 * 提醒时间（如任务未处理，啥时候提醒）
	 */
    private String reminderTime;
    /**
	 * 提醒间隔(分钟)（如任务未处理，提醒规则是什么）
	 */
    private Integer reminderRepeat;
    /**
	 * 是否自动执行（如任务未处理且到期，是否自动执行）
	 */
    private Boolean autoExecute;
}
  • EndModel
1
2
3
4
5
复制代码package com.mole.modules.flow.model;
import java.io.Serializable;
public class EndModel extends BaseModel implements Serializable {
    
}

流程节点解析类

将流程定义json转成对应的流程模型

  • FlowNodeParser.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
复制代码package com.mole.modules.flow.parser;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mole.modules.framework.util.JsonUtil;
import com.mole.modules.flow.model.BaseModel;
import com.mole.modules.flow.model.EndModel;
import com.mole.modules.flow.model.FlowModel;
import com.mole.modules.flow.model.StartModel;
import com.mole.modules.flow.model.TaskModel;
/**
 * 节点解析器
 * @author MLD
 *
 */
public class FlowNodeParser {
	private ObjectMapper objectMapper = new ObjectMapper();
	private final static String FLOW_ID="flowId";
	private final static String FLOW_NAME="flowName";
	private final static String NODE_LIST="nodeList";
	private final static String NODE_TYPE="nodeType";
	/**
	 * 解析流程定义
	 * @param flowDefine
	 * @return
	 * @throws IOException
	 */
	public FlowModel parser(String flowDefine) throws IOException {
		JsonNode root = objectMapper.readTree(flowDefine);
		if(!root.has(FLOW_ID)||!root.has(FLOW_NAME)||!root.has(NODE_LIST)) {
			
		}
		String flowId = root.get(FLOW_ID).asText();
		String flowName = root.get(FLOW_NAME).asText();
		JsonNode nodeList = root.get("nodeList");
		if(!nodeList.isArray()) {
			return null;
		}
		FlowModel flowModel = new FlowModel();
		flowModel.setFlowId(flowId);
		flowModel.setFlowName(flowName);
		flowModel.setNodeList(new ArrayList<BaseModel>());
		Iterator<JsonNode> nodes = nodeList.elements();
		while (nodes.hasNext()) {
			JsonNode node = nodes.next();
			if(node.has(NODE_TYPE)) {
				String nodeType = node.get(NODE_TYPE).asText();
				if("START".equals(nodeType)) {
					// 开始节点
					StartModel model = objectMapper.readValue(node.toString(), StartModel.class);
					flowModel.getNodeList().add(model);
				} else if("TASK".equals(nodeType)) {
					// 任务节点
					TaskModel model = objectMapper.readValue(node.toString(), TaskModel.class);
					flowModel.getNodeList().add(model);
				}	else if("END".equals(nodeType)) {
					// 结束节点
					EndModel model = objectMapper.readValue(node.toString(), EndModel.class);
					flowModel.getNodeList().add(model);
				}
			}
		}
		return flowModel;
	}
}

节点处理器类

节点处理器主要是对不同的节点进行不一样调度处理，如开始节点会驱动流程往下一步走，任务节点会产生任务，结束节点会将流程结束等。

  • FlowNodeProcessor.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
复制代码package com.mole.modules.flow.processor;

import com.mole.modules.flow.entity.FlowWork;
import com.mole.modules.flow.model.BaseModel;
import com.mole.modules.flow.model.FlowModel;
import com.mole.modules.flow.param.FlowParam;

/**
 * 节点处理器接口
 * @author MLD
 *
 */
public interface FlowNodeProcessor {
	/**
	 * 要处理的节点类型
	 * @return
	 */
	public String getNodeType();
	/**
	 * 流程节点处理方法
	 * @param flowWork 流程实例
	 * @param flowModel 当前流程模型
	 * @param currentNodeModel 当前节点模型
	 * @param flowParam 流程参数
	 */
	public void process(FlowWork flowWork,FlowModel flowModel,BaseModel currentNddeModel,FlowParam flowParam);
}
  • StartNodeProcessorImpl.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
复制代码package com.mole.modules.flow.processor.impl;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.mole.modules.flow.entity.FlowWork;
import com.mole.modules.flow.event.FlowStartNodeEvent;
import com.mole.modules.flow.model.BaseModel;
import com.mole.modules.flow.model.FlowModel;
import com.mole.modules.flow.model.StartModel;
import com.mole.modules.flow.param.FlowParam;
import com.mole.modules.flow.processor.FlowNodeProcessor;
/**
 * 开始节点处理器
 * @author MLD
 *
 */
@Component
public class StartNodeProcessorImpl implements FlowNodeProcessor {
	@Autowired(required=false)
	private FlowStartNodeEvent flowStartEvent;
	@Override
	public void process(FlowWork flowWork, FlowModel flowModel, BaseModel currentNddeModel,FlowParam flowParam) {
		// 开始节点事件--
		if(null != flowStartEvent) {
			StartModel startModel = flowModel.getStartModel();
			flowStartEvent.onEvent(flowWork, startModel, flowParam);
		}
	}
	@Override
	public String getNodeType() {
		return "START";
	}

}
  • TaskNodeProcessorImpl.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
复制代码package com.mole.modules.flow.processor.impl;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.mole.framework.base.ServiceException;
import com.mole.framework.base.YesNoEnum;
import com.mole.framework.util.JsonUtil;
import com.mole.framework.util.StringUtil;
import com.mole.modules.flow.entity.FlowTask;
import com.mole.modules.flow.entity.FlowWork;
import com.mole.modules.flow.enums.FlowErrEnum;
import com.mole.modules.flow.event.FlowTaskCreatedEvent;
import com.mole.modules.flow.event.FlowTaskNodeEvent;
import com.mole.modules.flow.mapper.FlowTaskMapper;
import com.mole.modules.flow.mapper.FlowWorkMapper;
import com.mole.modules.flow.model.BaseModel;
import com.mole.modules.flow.model.FlowModel;
import com.mole.modules.flow.model.TaskModel;
import com.mole.modules.flow.param.FlowParam;
import com.mole.modules.flow.processor.FlowNodeProcessor;
import com.mole.modules.flow.service.FlowNodeUserService;
/**
 * 任务节点处理器
 * @author MLD
 *
 */
@Component
public class TaskNodeProcessorImpl implements FlowNodeProcessor {
	@Autowired
	private FlowWorkMapper flowWorkMapper;
	@Autowired
	private FlowTaskMapper flowTaskMapper;
	@Autowired(required=false)
	private FlowNodeUserService flowNodeUserService;
	@Autowired(required=false)
	private FlowTaskNodeEvent flowTaskStartEvent;
	@Autowired(required=false)
	private FlowTaskCreatedEvent flowTaskCreatedEvent;
	@Override
	public void process(FlowWork flowWork, FlowModel flowModel, BaseModel currentNddeModel,FlowParam flowParam) {
		// 下一个节点
		// BaseModel nextNodeModel = flowModel.getNodeModel(flowWork.getNextNodeId());
		TaskModel taskModel = (TaskModel)currentNddeModel;
		// 1. 创建任务
		createTask(flowWork, flowParam, taskModel);
		// 2. 修改当前流程实例状态
		if(null != flowTaskStartEvent) {
			flowTaskStartEvent.OnEvent(flowWork, taskModel, flowParam);
		}
	}
	/**
	 * 创建任务
	 * @param flowWork
	 * @param param
	 * @param taskModel
	 */
	private int createTask(FlowWork flowWork,FlowParam flowParam,TaskModel taskModel){
		Date now = new Date();
		String flowWorkId = flowWork.getId();
		List<String> userIds = flowParam.getUserIds();
		if(null==userIds) {
			userIds = new ArrayList<>();
		}
		if(null!=flowNodeUserService) {
			List<String> list = flowNodeUserService.loadUser(taskModel, flowParam);
			if(null!=list) {
				for(String id:list) {
					if(!userIds.contains(id)) {
						userIds.add(id);
					}
				}
				
			}
		}
		if(userIds.isEmpty()) {
			throw new ServiceException(FlowErrEnum.FLOW86000009);
		}
		for(String actorUserId:userIds) {
			// 2. 创建一个任务
			FlowTask flowTask = new FlowTask();
			flowTask.setCreateTime(now);
			flowTask.setFlowWorkId(flowWorkId);
			flowTask.setIsDeleted(YesNoEnum.NO);
			flowTask.setServiceId(flowParam.getServiceId());
			flowTask.setServiceType(flowParam.getServiceType());
			flowTask.setTaskName(taskModel.getName());
			flowTask.setUpdateTime(now);
			flowTask.setOperator(flowParam.getOperator());
			flowTask.setStatus(FlowTask.StatusEnum.CREATED);
			flowTask.setFlowNodeId(taskModel.getId());
			flowTask.setActorUserId(actorUserId);
			flowTask.setFlowParam(JsonUtil.toJson(flowParam));
			if(StringUtil.isNotEmpty(flowParam.getTitle())) {
				flowTask.setTaskName(flowParam.getTitle());
			}
			flowTaskMapper.insertSelective(flowTask);
			if(null != flowTaskCreatedEvent) {
				flowTaskCreatedEvent.onEvent(flowWork, flowTask, flowParam);
			}
		}
		return userIds.size();
	}

	@Override
	public String getNodeType() {
		return "TASK";
	}

}
  • EndNodeProcessorImpl.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
复制代码package com.mole.modules.flow.processor.impl;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.mole.modules.flow.entity.FlowWork;
import com.mole.modules.flow.event.FlowEndNodeEvent;
import com.mole.modules.flow.model.BaseModel;
import com.mole.modules.flow.model.EndModel;
import com.mole.modules.flow.model.FlowModel;
import com.mole.modules.flow.param.FlowParam;
import com.mole.modules.flow.processor.FlowNodeProcessor;
@Component
public class EndNodeProcessorImpl implements FlowNodeProcessor{
	@Autowired(required=false)
	private FlowEndNodeEvent flowEndEvent;
	@Override
	public void process(FlowWork flowWork, FlowModel flowModel, BaseModel currentNddeModel,FlowParam flowParam) {
		if(null == flowEndEvent) {
			EndModel endModel = flowModel.getEndModel();
			flowEndEvent.onEvent(flowWork, endModel, flowParam);;
		}
	}

	@Override
	public String getNodeType() {
		return "END";
	}

}

节点处理器控制类

由节点处理器控制类统一调用处理器

  • FlowNodeProcesstorCtl.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
复制代码package com.mole.modules.flow.processor;

import java.util.Date;
import java.util.List;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.mole.modules.flow.entity.FlowWork;
import com.mole.modules.flow.mapper.FlowWorkMapper;
import com.mole.modules.flow.model.BaseModel;
import com.mole.modules.flow.model.EndModel;
import com.mole.modules.flow.model.FlowModel;
import com.mole.modules.flow.model.StartModel;
import com.mole.modules.flow.param.FlowParam;
/**
 * 节点处理器控制器
 * @author MLD
 *
 */
@Component
public class FlowNodeProcesstorCtl {
	@Autowired(required=false)
	private List<FlowNodeProcessor> flowNodeProcessorList;
	@Autowired
	private FlowWorkMapper flowWorkMapper;
	/**
	 * 节点流程处理器控制
	 * @param flowWork 当前流程实例
	 * @param flowModel	当前流程实例模型
	 * @param model 当前节点模型
	 * @param param 流程参数
	 */
	public void process(FlowWork flowWork,FlowModel flowModel, BaseModel model,FlowParam param){
		FlowWork upFlowWork = new FlowWork();
		if(model == null) {
			return;
		}
		// 执行开始节点
		for(FlowNodeProcessor processor:flowNodeProcessorList) {
			if(processor.getNodeType().equals(model.getNodeType())){
				BaseModel nextNodeModel = flowModel.getNodeModel(model.getNextNodeId());
				if(model instanceof StartModel) { //开始节点
					// 执行开始模型
					processor.process(flowWork, flowModel, model,param);
					// 执行下一个模型
					process(flowWork,flowModel,nextNodeModel, param);
				} else if(model instanceof EndModel)  {
					// 结束节点
					upFlowWork.setStatus(FlowWork.StatusEnum.END);
					Date now = new Date();
					upFlowWork.setId(flowWork.getId());
					upFlowWork.setCurrentNodeId(model.getId());
					upFlowWork.setNextNodeId("");
					upFlowWork.setLastOperator(param.getOperator());
					upFlowWork.setUpdateTime(now);
					flowWorkMapper.updateByPrimaryKeySelective(upFlowWork);
				} else { // 非开始和结束节点
					processor.process(flowWork, flowModel, model,param);
					Date now = new Date();
					upFlowWork.setId(flowWork.getId());
					upFlowWork.setCurrentNodeId(model.getId());
					upFlowWork.setNextNodeId(model.getNextNodeId());
					upFlowWork.setLastOperator(param.getOperator());
					upFlowWork.setUpdateTime(now);
					flowWorkMapper.updateByPrimaryKeySelective(upFlowWork);
				}
				// 存在一个满足条件的，即可中断，其实设计上也仅会有一个相同类型的。
				break;
			}
		}
	}
}

节点事件接口类

节点事件接口，由业务系统实现

  • FlowStartNodeEvent
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
复制代码package com.mole.modules.flow.event;

import com.mole.modules.flow.entity.FlowWork;
import com.mole.modules.flow.model.StartModel;
import com.mole.modules.flow.param.FlowParam;
/**
 * 流程实例开始事件-由业务系统实现
 * @author MLD
 *
 */
public interface FlowStartNodeEvent {
	/**
	 * @param flowWork 当前流程实例
	 * @param startModel 开始节点模型
	 * @param param 流程参数
	 */
	public void onEvent(FlowWork flowWork, StartModel startModel,FlowParam flowParam);
}
  • FlowTaskNodeEvent.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
复制代码package com.mole.modules.flow.event;

import com.mole.modules.flow.entity.FlowWork;
import com.mole.modules.flow.model.TaskModel;
import com.mole.modules.flow.param.FlowParam;

/**
 * 任务开始事件-由业务系统实现
 * @author MLD
 *
 */
public interface FlowTaskNodeEvent {
	/**
	 * 任务开始事件
	 * @param flowWork 当前流程实例
	 * @param taskModel 任务节点模型
	 * @param flowParam 流程参数
	 */
	public void OnEvent(FlowWork flowWork,TaskModel taskModel,FlowParam flowParam);
}
  • FlowTaskCreatedEvent.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
复制代码package com.mole.modules.flow.event;

import com.mole.modules.flow.entity.FlowTask;
import com.mole.modules.flow.entity.FlowWork;
import com.mole.modules.flow.param.FlowParam;

/**
 * 创建任务事件-由业务系统实现
 * @author MLD
 *
 */
public interface FlowTaskCreatedEvent {
	/**
	 * 
	 * @param flowWork 当前流程实例
	 * @param flowTask 当前任务实例
	 * @param flowParam 流程参数
	 */
	public void onEvent(FlowWork flowWork,FlowTask flowTask,FlowParam flowParam);
}
  • FlowEndNodeEvent.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
复制代码package com.mole.modules.flow.event;

import com.mole.modules.flow.entity.FlowWork;
import com.mole.modules.flow.model.EndModel;
import com.mole.modules.flow.param.FlowParam;
/**
 * 流程实例结束事件-由业务系统实现
 * @author MLD
 *
 */
public interface FlowEndNodeEvent {
	/**
	 * 
	 * @param flowWork 当前流程实例
	 * @param endModel 结束节点
	 * @param flowParam 流程参数
	 */
	public void onEvent(FlowWork flowWork,EndModel endModel, FlowParam flowParam);
}

自定义任务节点参与人接口类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
复制代码package com.mole.modules.flow.service;

import java.util.List;

import com.mole.modules.flow.model.BaseModel;
import com.mole.modules.flow.param.FlowParam;
/**
 * 自定义节点参与人接口-由业务系统实现
 * @author MLD
 *
 */
public interface FlowNodeUserService {
	/**
	 * 加载节点参与人
	 * @param nodeModel 节点模型
	 * @param param 流程参数
	 * @return
	 */
	public List<String> loadUser(BaseModel nodeModel,FlowParam param);
}

流程控制类

提供给业务操作流程的类

  • FlowWorkCtl.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
复制代码package com.mole.modules.flow;

import java.io.IOException;
import java.util.Date;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import tk.mybatis.mapper.entity.Condition;

import com.mole.framework.base.ServiceException;
import com.mole.framework.base.YesNoEnum;
import com.mole.framework.util.JsonUtil;
import com.mole.modules.flow.entity.FlowDefine;
import com.mole.modules.flow.entity.FlowTask;
import com.mole.modules.flow.entity.FlowWork;
import com.mole.modules.flow.enums.FlowErrEnum;
import com.mole.modules.flow.mapper.FlowDefineMapper;
import com.mole.modules.flow.mapper.FlowTaskMapper;
import com.mole.modules.flow.mapper.FlowWorkMapper;
import com.mole.modules.flow.model.BaseModel;
import com.mole.modules.flow.model.FlowModel;
import com.mole.modules.flow.model.StartModel;
import com.mole.modules.flow.param.FlowParam;
import com.mole.modules.flow.parser.FlowNodeParser;
import com.mole.modules.flow.processor.FlowNodeProcesstorCtl;

/**
 * 流程控制类-引擎
 * @author MLD
 *
 */
@Component
public class FlowWorkCtl {
    /**
	 * 通过流程定义编码启动流程
	 * @param flowDefineCode 流程定义编码
	 * @param param 流程启动入参
	 */
	@Autowired
	private FlowDefineMapper flowDefineMapper;
	@Autowired
	private FlowNodeParser flowNodeParser;
	@Autowired
	private FlowWorkMapper flowWorkMapper;
	@Autowired
	private FlowNodeProcesstorCtl flowNodeProcesstorCtl;
	@Autowired
	private FlowTaskMapper flowTaskMapper;
	/**
	 * 启动一个流程
	 * @param flowDefineCode
	 * @param operator 操作者用户id
	 * @param param
	 */
    public FlowWork startProcess(String flowDefineCode,String operator,FlowParam param){
        // TODO 加载流程配置
    	FlowDefine q = new FlowDefine();
    	q.setFlowDefineCode(flowDefineCode);
    	q.setIsDeleted(YesNoEnum.NO);
    	if(null==param.getOperator()) {
    		param.setOperator(operator);
    	}
    	FlowDefine flowDefine = flowDefineMapper.selectOne(q);
    	if(null == flowDefine) {
    		// 流程定义不存在
    		throw new ServiceException(FlowErrEnum.FLOW86000001);
    	}
    	FlowModel flowModel = null;
    	try {
    		flowModel = flowNodeParser.parser(flowDefine.getFlowDefine());
		} catch (IOException e) {
			// 流程解析异常
    		throw new ServiceException(FlowErrEnum.FLOW86000002);
		}
    	StartModel startModel = flowModel.getStartModel();
    	Date now = new Date();
    	// 创建一个新流程
		FlowWork flowWork = new FlowWork();
		flowWork.setCreateTime(now);
		flowWork.setCurrentNodeId(startModel.getId());
		flowWork.setFlowDefine(flowDefine.getFlowDefine());
		flowWork.setFlowName(flowDefine.getFlowName());
		flowWork.setIsDeleted(YesNoEnum.NO);
		flowWork.setNextNodeId(startModel.getNextNodeId());
		flowWork.setStatus(FlowWork.StatusEnum.START);
		flowWork.setUpdateTime(now);
		flowWork.setFlowParam(JsonUtil.toJson(param));
		flowWork.setFlowDefineCode(flowDefineCode);
		flowWork.setLastOperator(operator);
		flowWorkMapper.insertSelective(flowWork);
		// 执行开始模型
		flowNodeProcesstorCtl.process(flowWork, flowModel, startModel, param);
		return flowWork;
    }
    /**
     * 完成一个任务
     * @param flowTaskId 流程任务id
     * @param operator 操作者用户id
     * @param param 流程参数
     */
    public void completeTask(String flowTaskId,String operator,FlowParam param) {
    	if(null==param.getOperator()) {
    		param.setOperator(operator);
    	}
    	FlowTask task = flowTaskMapper.selectByPrimaryKey(flowTaskId);
		if(null == task || YesNoEnum.YES.equals(task.getIsDeleted())) {
			// 流程任务不存在
			throw new ServiceException(FlowErrEnum.FLOW86000007);
		}
		// 判断操作者是否在分派的用户里
		if(!task.getActorUserId().equals(operator)) {
			// 该用户无流程任务
			throw new ServiceException(FlowErrEnum.FLOW86000008);
		}
		if(FlowTask.StatusEnum.FINISHED.equals(task.getStatus())) {
			// 已完成，就不处理了
			return;
		}
		FlowWork flowWork = flowWorkMapper.selectByPrimaryKey(task.getFlowWorkId());
		if(null==flowWork) {
			// 流程实例不存在
			throw new ServiceException(FlowErrEnum.FLOW86000005);
		}
		FlowModel flowModel = null;
    	try {
    		flowModel = flowNodeParser.parser(flowWork.getFlowDefine());
		} catch (IOException e) {
			// 流程解析异常
    		throw new ServiceException(FlowErrEnum.FLOW86000002);
		}
		BaseModel currentNodeModel = flowModel.getNodeModel(task.getFlowNodeId());
    	BaseModel nextNodeModel = flowModel.getNodeModel(currentNodeModel.getNextNodeId());
		Date now = new Date();
		FlowTask upTask = new FlowTask();
		upTask.setId(task.getId());
		upTask.setUpdateTime(now);
		upTask.setStatus(FlowTask.StatusEnum.FINISHED);
		upTask.setFinishTime(now);
		// 修改当前任务为已完成
		flowTaskMapper.updateByPrimaryKeySelective(upTask);
		int count = 0;
		if("ALL".equals(currentNodeModel.getNodeType())) {
			// 所有人都完成，才能走下一步流程
			Condition qTaskCondition = new Condition(FlowTask.class);
			qTaskCondition.createCriteria().andEqualTo("flowWorkId", task.getFlowWorkId())
			.andEqualTo("flowNodeId", task.getFlowNodeId())
			.andEqualTo("status", FlowTask.StatusEnum.CREATED)
			.andEqualTo("isDeleted", YesNoEnum.NO)
			.andNotEqualTo("actorUserId", operator);
			count = flowTaskMapper.selectCountByCondition(qTaskCondition);
		} else {
			// ANY 任务一个节点完成都可以走下一步，别的任务要修改为已取消
			Condition upTaskCondition = new Condition(FlowTask.class);
			upTaskCondition.createCriteria().andEqualTo("flowWorkId", task.getFlowWorkId())
			.andEqualTo("flowNodeId", task.getFlowNodeId())
			.andEqualTo("status", FlowTask.StatusEnum.CREATED)
			.andEqualTo("isDeleted", YesNoEnum.NO)
			.andNotEqualTo("actorUserId", operator);
			FlowTask upTaskToCancel = new FlowTask();
			upTaskToCancel.setUpdateTime(now);
			upTaskToCancel.setStatus(FlowTask.StatusEnum.CANCEL);
			flowTaskMapper.updateByPrimaryKeySelective(upTaskToCancel);
		}
		if (count==0) {
			// 3. 执行下一步流程
			flowNodeProcesstorCtl.process(flowWork, flowModel, nextNodeModel, param);
		}
    }
    /**
     * 驳回一个任务
     * @param flowTaskId 流程任务id
     * @param operator 操作者用户id
     * @param flowParam 流程参数
     */
    public void rejectTask(String flowTaskId,String operator,FlowParam flowParam) {
    	if(null==flowParam.getOperator()) {
    		flowParam.setOperator(operator);
    	}
    	FlowTask task = flowTaskMapper.selectByPrimaryKey(flowTaskId);
		if(null == task || YesNoEnum.YES.equals(task.getIsDeleted())) {
			// 流程任务不存在
			throw new ServiceException(FlowErrEnum.FLOW86000007);
		}
		// 判断人是否在里面
		if(!task.getActorUserId().equals(operator)) {
			// 该用户无流程任务
			throw new ServiceException(FlowErrEnum.FLOW86000008);
		}
		if(FlowTask.StatusEnum.FINISHED.equals(task.getStatus())) {
			// 已完成，就不处理了
			return;
		}
		FlowWork flowWork = flowWorkMapper.selectByPrimaryKey(task.getFlowWorkId());
		if(null==flowWork) {
			// 流程实例不存在
			throw new ServiceException(FlowErrEnum.FLOW86000005);
		}
		FlowModel flowModel = null;
    	try {
    		flowModel = flowNodeParser.parser(flowWork.getFlowDefine());
		} catch (IOException e) {
			// 流程解析异常
    		throw new ServiceException(FlowErrEnum.FLOW86000002);
		}
		Date now = new Date();
		FlowTask upTask = new FlowTask();
		upTask.setId(task.getId());
		upTask.setUpdateTime(now);
		upTask.setStatus(FlowTask.StatusEnum.REJECT);
		upTask.setFinishTime(now);
		// 修改当前任务为驳回
		flowTaskMapper.updateByPrimaryKeySelective(upTask);
		// 中断当前节点任务
		Condition upTaskCondition = new Condition(FlowTask.class);
		upTaskCondition.createCriteria().andEqualTo("flowWorkId", task.getFlowWorkId())
		.andEqualTo("flowNodeId", task.getFlowNodeId())
		.andEqualTo("status", FlowTask.StatusEnum.CREATED)
		.andEqualTo("isDeleted", YesNoEnum.NO)
		.andNotEqualTo("actorUserId", operator);
		FlowTask upTaskToCancel = new FlowTask();
		upTaskToCancel.setUpdateTime(now);
		upTaskToCancel.setStatus(FlowTask.StatusEnum.CANCEL);
		flowTaskMapper.updateByPrimaryKeySelective(upTaskToCancel);
		
		// 获取上一个节点
    	BaseModel preNodeModel = flowModel.getPreNodeModel(task.getFlowNodeId());
    	// 执行上一个节点处理器
    	flowNodeProcesstorCtl.process(flowWork, flowModel, preNodeModel, flowParam);
    }
}

调用样例

  • 启动一个流程实例
1
2
3
4
5
6
7
8
复制代码String flowDefineCode = "yzq_invite_bid";
FlowParam flowParam = new FlowParam();
flowParam.setServiceId(UUID.randomUUID().toString());
flowParam.setServiceType("yzq_invite_bid");
flowParam.setUserIds(new ArrayList<>());
flowParam.getUserIds().add("1");
Sting operator = "1";
flowWorkCtl.startProcess(flowDefineCode,operator, flowParam);
  • 完成一个任务
1
2
3
4
5
6
7
8
复制代码String flowTaskId = "flowTaskId";
FlowParam flowParam = new FlowParam();
flowParam.setServiceId(UUID.randomUUID().toString());
flowParam.setServiceType("yzq_invite_bid");
flowParam.setUserIds(new ArrayList<>());
flowParam.getUserIds().add("1");
Sting operator = "1";
flowWorkCtl.completeTask(flowTaskId, operator, flowParam);
  • 驳回任务
1
2
3
4
5
6
7
8
复制代码String flowTaskId = "flowTaskId";
FlowParam flowParam = new FlowParam();
flowParam.setServiceId(UUID.randomUUID().toString());
flowParam.setServiceType("yzq_invite_bid");
flowParam.setUserIds(new ArrayList<>());
flowParam.getUserIds().add("1");
Sting operator = "1";
flowWorkCtl.rejectTask(flowTaskId, operator, flowParam);

本工程使用到的框架

  • springboot 2.0
  • tk.mybatis

其他

本想开源的，但是该项目和公司的框架耦合的太厉害，不太方便抽离，等哪天有空，会考虑把这部分代码单独抽离开源。还有一点就是，目前考虑的只是最简单的顺序流程，涉及到条件、分支、合并、子流程等复杂的流程暂时还没考虑。不过本意只是想一步步来，慢慢剖析原理。

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%