PSI 进销存系统工作流引擎设计与流程自动化实现
引言
工作流引擎是进销存系统的核心组件，负责管理采购审批、销售审批、费用报销等业务流程。本文将介绍 PSI 进销存系统中工作流引擎的设计与实现，支持业务流程自定义配置。
工作流架构
工作流引擎整体架构：
| 组件 | 职责 | 技术实现 |
|---|---|---|
| 流程设计器 | 可视化流程配置 | 前端组件 |
| 流程引擎 | 流程执行与调度 | 状态机 |
| 任务分配器 | 待办任务分配 | 规则引擎 |
| 通知服务 | 消息推送 | 消息队列 |
核心代码实现
工作流引擎实现：
// 工作流引擎
class WorkflowEngine {
constructor(config) {
this.processDefinitions = new Map();
this.taskService = config.taskService;
this.notificationService = config.notificationService;
this.historyService = config.historyService;
}
// 部署流程定义
async deployProcess(processDef) {
const process = new ProcessDefinition(processDef);
this.processDefinitions.set(process.id, process);
return process;
}
// 启动流程实例
async startProcess(processDefId, businessKey, initiator, variables = {}) {
const processDef = this.processDefinitions.get(processDefId);
if (!processDef) {
throw new Error(`Process definition not found: ${processDefId}`);
}
// 创建流程实例
const processInstance = new ProcessInstance({
processDefId,
businessKey,
initiator,
status: 'running',
variables: { ...variables }
});
// 执行开始节点
const startNode = processDef.getStartNode();
processInstance.currentNode = startNode.id;
// 执行流转
await this.executeNode(processInstance, startNode);
return processInstance;
}
// 执行节点
async executeNode(processInstance, node) {
// 记录节点进入历史
await this.historyService.recordNodeEntry(processInstance, node);
switch (node.type) {
case 'start':
// 查找下一步
const outgoing = processDef.getOutgoingFlows(node.id);
for (const flow of outgoing) {
if (await this.evaluateCondition(processInstance, flow.condition)) {
const nextNode = processDef.getNode(flow.targetRef);
await this.executeNode(processInstance, nextNode);
}
}
break;
case 'userTask':
// 创建用户任务
await this.createUserTask(processInstance, node);
break;
case 'serviceTask':
// 执行服务任务
await this.executeServiceTask(processInstance, node);
break;
case 'exclusiveGateway':
// 排他网关
const conditions = processDef.getOutgoingFlows(node.id);
for (const flow of conditions) {
if (await this.evaluateCondition(processInstance, flow.condition)) {
const nextNode = processDef.getNode(flow.targetRef);
await this.executeNode(processInstance, nextNode);
return;
}
}
break;
case 'parallelGateway':
// 并行网关 - 开启所有分支
const parallelFlows = processDef.getOutgoingFlows(node.id);
await Promise.all(parallelFlows.map(async (flow) => {
const nextNode = processDef.getNode(flow.targetRef);
await this.executeNode(processInstance, nextNode);
}));
break;
case 'end':
// 流程结束
processInstance.status = 'completed';
processInstance.endTime = new Date();
await this.historyService.recordProcessEnd(processInstance);
break;
}
}
// 创建用户任务
async createUserTask(processInstance, node) {
// 获取任务分配
const candidateUsers = await this.resolveCandidateUsers(
processInstance, node.candidateUsers
);
const candidateGroups = await this.resolveCandidateGroups(
processInstance, node.candidateGroups
);
const task = {
id: this.generateId(),
processInstanceId: processInstance.id,
nodeId: node.id,
name: node.name,
description: node.documentation,
status: 'pending',
candidateUsers,
candidateGroups,
createTime: new Date(),
dueDate: node.dueDate ?
this.calculateDueDate(node.dueDate, processInstance) : null
};
await this.taskService.createTask(task);
// 发送通知
await this.notificationService.notifyUsers(
candidateUsers,
'您有新的待办任务',
{ taskId: task.id, taskName: task.name }
);
}
// 完成任务
async completeTask(taskId, userId, variables = {}) {
const task = await this.taskService.getTask(taskId);
if (!task) {
throw new Error(`Task not found: ${taskId}`);
}
// 验证任务权限
if (!this.hasTaskPermission(task, userId)) {
throw new Error('User does not have permission to complete this task');
}
// 更新任务状态
task.status = 'completed';
task.completedBy = userId;
task.completedTime = new Date();
task.variables = variables;
await this.taskService.updateTask(task);
// 更新流程实例变量
const processInstance = await this.getProcessInstance(task.processInstanceId);
Object.assign(processInstance.variables, variables);
// 继续执行流程
const processDef = this.processDefinitions.get(processInstance.processDefId);
const currentNode = processDef.getNode(task.nodeId);
const outgoingFlows = processDef.getOutgoingFlows(currentNode.id);
for (const flow of outgoingFlows) {
if (!flow.condition || await this.evaluateCondition(processInstance, flow.condition)) {
const nextNode = processDef.getNode(flow.targetRef);
await this.executeNode(processInstance, nextNode);
return;
}
}
}
// 条件评估
async evaluateCondition(processInstance, condition) {
if (!condition) return true;
// 简单表达式评估
const vars = processInstance.variables;
try {
// 安全评估：只允许访问预定义的变量
const safeVars = {
amount: vars.amount || 0,
quantity: vars.quantity || 0,
price: vars.price || 0,
total: vars.total || 0,
status: vars.status || '',
type: vars.type || '',
customerId: vars.customerId || '',
userId: vars.userId || '',
department: vars.department || ''
};
// 使用 Function 构造函数（实际生产应使用安全的表达式解析器）
const fn = new Function('vars', `with(vars) { return ${condition}; }`);
return fn(safeVars);
} catch (error) {
console.error('Condition evaluation failed:', error);
return false;
}
}
// 获取任务分配
async resolveCandidateUsers(processInstance, candidateUsers) {
if (!candidateUsers) return [];
if (typeof candidateUsers === 'string') {
return [candidateUsers];
}
if (Array.isArray(candidateUsers)) {
return candidateUsers;
}
if (candidateUsers.type === 'initiator') {
return [processInstance.initiator];
}
if (candidateUsers.type === 'variable') {
return [processInstance.variables[candidateUsers.expression]];
}
if (candidateUsers.type === 'script') {
const fn = new Function('vars', `with(vars) { return ${candidateUsers.expression}; }`);
return fn(processInstance.variables);
}
return [];
}
generateId() {
return `${Date.now()}-${Math.random().toString(36).substr(2, 9)}`;
}
}
// 流程定义
class ProcessDefinition {
constructor(config) {
this.id = config.id || this.generateId();
this.name = config.name;
this.version = config.version || 1;
this.nodes = new Map();
this.flows = [];
// 解析节点
config.nodes.forEach(node => {
this.nodes.set(node.id, node);
});
// 解析连线
this.flows = config.flows || [];
}
getStartNode() {
return [...this.nodes.values()].find(n => n.type === 'start');
}
getNode(nodeId) {
return this.nodes.get(nodeId);
}
getOutgoingFlows(nodeId) {
return this.flows.filter(f => f.sourceRef === nodeId);
}
generateId() {
return `process_${Date.now()}`;
}
}
// 流程实例
class ProcessInstance {
constructor(config) {
this.id = this.generateId();
this.processDefId = config.processDefId;
this.businessKey = config.businessKey;
this.initiator = config.initiator;
this.status = config.status || 'running';
this.variables = config.variables || {};
this.currentNode = null;
this.startTime = new Date();
this.endTime = null;
}
generateId() {
return `instance_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
}
}
流程设计器
可视化流程配置：
// 流程设计器数据模型
class ProcessDesigner {
// 导出流程定义 JSON
exportProcess(nodes, flows) {
const processDef = {
id: `process_${Date.now()}`,
name: '新流程',
version: 1,
nodes: nodes.map(node => this.exportNode(node)),
flows: flows.map(flow => this.exportFlow(flow))
};
return processDef;
}
// 导出节点
exportNode(node) {
return {
id: node.id,
name: node.name,
type: node.type,
documentation: node.documentation || '',
x: node.x,
y: node.y,
// 用户任务特定属性
candidateUsers: node.candidateUsers || null,
candidateGroups: node.candidateGroups || null,
dueDate: node.dueDate || null,
// 服务任务特定属性
serviceType: node.serviceType || null,
serviceConfig: node.serviceConfig || null,
// 网关特定属性
defaultFlow: node.defaultFlow || null
};
}
// 导出连线
exportFlow(flow) {
return {
id: flow.id,
sourceRef: flow.source.id,
targetRef: flow.target.id,
condition: flow.condition || null,
name: flow.name || ''
};
}
// 导入流程定义
importProcess(json) {
return {
nodes: json.nodes.map(n => this.importNode(n)),
flows: json.flows.map(f => this.importFlow(f, json))
};
}
// 验证流程定义
validateProcess(nodes, flows) {
const errors = [];
// 检查是否包含开始节点
const startNodes = nodes.filter(n => n.type === 'start');
if (startNodes.length === 0) {
errors.push('流程必须包含开始节点');
}
// 检查是否有未连接的节点
const connectedNodes = new Set();
flows.forEach(f => {
connectedNodes.add(f.source.id);
connectedNodes.add(f.target.id);
});
nodes.forEach(node => {
if (!connectedNodes.has(node.id) && node.type !== 'start' && node.type !== 'end') {
errors.push(`节点 "${node.name}" 未连接到流程`);
}
});
// 检查结束节点
const endNodes = nodes.filter(n => n.type === 'end');
if (endNodes.length === 0) {
errors.push('流程必须包含结束节点');
}
// 检查死循环
if (this.hasDeadLoop(nodes, flows)) {
errors.push('流程存在死循环');
}
return {
valid: errors.length === 0,
errors
};
}
// 检查死循环
hasDeadLoop(nodes, flows) {
// 简化的死循环检测
const inDegree = new Map();
const outDegree = new Map();
nodes.forEach(n => {
inDegree.set(n.id, 0);
outDegree.set(n.id, 0);
});
flows.forEach(f => {
inDegree.set(f.target.id, inDegree.get(f.target.id) + 1);
outDegree.set(f.source.id, outDegree.get(f.source.id) + 1);
});
// 检查是否有节点只有出度没有入度（除了开始节点）
for (const [nodeId, inD] of inDegree) {
const outD = outDegree.get(nodeId);
const node = nodes.find(n => n.id === nodeId);
if (inD === 0 && outD > 0 && node.type !== 'start') {
return true;
}
}
return false;
}
}
// 常用流程模板
const ProcessTemplates = {
// 采购审批流程
purchaseApproval: {
name: '采购审批流程',
nodes: [
{ id: 'start', name: '开始', type: 'start', x: 100, y: 200 },
{ id: 'apply', name: '提交申请', type: 'userTask', x: 250, y: 200 },
{ id: 'gateway1', name: '金额判断', type: 'exclusiveGateway', x: 400, y: 200 },
{ id: 'approve1', name: '部门经理审批', type: 'userTask', x: 550, y: 100 },
{ id: 'approve2', name: '总经理审批', type: 'userTask', x: 550, y: 300 },
{ id: 'execute', name: '执行采购', type: 'userTask', x: 700, y: 200 },
{ id: 'end', name: '结束', type: 'end', x: 850, y: 200 }
],
flows: [
{ source: { id: 'start' }, target: { id: 'apply' } },
{ source: { id: 'apply' }, target: { id: 'gateway1' } },
{ source: { id: 'gateway1' }, target: { id: 'approve1' }, condition: 'amount <= 10000' },
{ source: { id: 'gateway1' }, target: { id: 'approve2' }, condition: 'amount > 10000' },
{ source: { id: 'approve1' }, target: { id: 'execute' } },
{ source: { id: 'approve2' }, target: { id: 'execute' } },
{ source: { id: 'execute' }, target: { id: 'end' } }
]
},
// 销售审批流程
salesApproval: {
name: '销售审批流程',
nodes: [
{ id: 'start', name: '开始', type: 'start', x: 100, y: 200 },
{ id: 'create', name: '创建订单', type: 'userTask', x: 250, y: 200 },
{ id: 'check', name: '库存检查', type: 'serviceTask', x: 400, y: 200 },
{ id: 'approve', name: '销售审批', type: 'userTask', x: 550, y: 200 },
{ id: 'delivery', name: '发货', type: 'userTask', x: 700, y: 200 },
{ id: 'end', name: '结束', type: 'end', x: 850, y: 200 }
],
flows: [
{ source: { id: 'start' }, target: { id: 'create' } },
{ source: { id: 'create' }, target: { id: 'check' } },
{ source: { id: 'check' }, target: { id: 'approve' } },
{ source: { id: 'approve' }, target: { id: 'delivery' } },
{ source: { id: 'delivery' }, target: { id: 'end' } }
]
}
};
最佳实践建议
- 流程建模：使用可视化设计器简化流程配置，降低使用门槛
- 权限控制：严格控制流程设计权限，防止恶意流程影响业务
- 超时处理：设置任务超时提醒和自动催办，提升流程效率
- 监控分析：记录流程执行时间，识别瓶颈优化流程
- 回滚机制：支持流程撤销和回退，处理异常情况
总结
PSI 进销存系统的工作流引擎提供了完整的流程自动化能力：
- 可视化配置：无需编码即可自定义业务流程
- 灵活分配：支持多种任务分配方式
- 条件流转：支持排他网关和并行网关
- 全程追踪：完整的执行历史记录
通过工作流引擎，可以显著提升进销存系统的业务处理效率和规范性。