技术学习分享_CKX技术 技术资讯 轻量级工作流引擎的设计与实现

轻量级工作流引擎的设计与实现

广告位

一、什么是工作流引擎

工作流引擎是驱动工作流执行的一套代码。

至于什么是工作流、为什么要有工作流、工作流的应用景,同学们可以看一看网上的资料,在此处不在展开。

 

二、为什么要重复造轮子

开源的工作流引擎很多,比如 activiti、flowable、Camunda 等,那么,为什么没有选它们呢?基于以下几点考虑:

  • 最重要的,满足不了业务需求,一些特殊的场景无法实现。
  • 有些需求实现起来比较绕,更有甚者,需要直接修改引擎数据库,这对于引擎的稳定运行带来了巨大的隐患,也对以后引擎的版本升级制造了一些困难。
  • 资料、代码量、API繁多,学习成本较高,维护性较差。
  • 经过分析与评估,我们的业务场景需要的BPMN元素较少,开发实现的代价不大。

因此,重复造了轮子,其实,还有一个更深层次的战略上的考虑,即:作为科技公司,我们一定要有我们自己的核心底层技术!这样,才能不受制于人(参考最近的芯片问题)。

 

三、怎么造的轮子

对于一次学习型分享来讲,过程比结果更重要,那些只说结果,不细说过程甚至不说的分享,我认为是秀肌肉,而不是真正意义上的分享。因此,接下来,本文将重点描述造轮子的主要过程。

一个成熟的工作流引擎的构建是很复杂的,如何应对这种复杂性呢?一般来讲,有以下三种方法:

  • 确定性交付:弄清楚需求是什么,验收标准是什么,最好能够写出测试用例,这一步是为了明确目标。
  • 迭代式开发:先从小的问题集的解决开始,逐步过渡到解决大的问题集上来,罗马不是一天建成的,人也不是一天就能成熟的,是需要个过程的。
  • 分而治之:把大的问题拆成小的问题,小问题的解决会推动大问题的解决(这个思想适用场景比较多,同学们可以用心体会和理解哈)。

如果按照上述方法,一步一步的详细展开,那么可能需要一本书。为了缩减篇幅而又不失干货,本文会描述重点几个迭代,进而阐述轻量级工作流引擎的设计与主要实现。

那么,轻量级又是指什么呢?这里,主要是指以下几点

  • 少依赖:代码的java实现上,除了jdk8以外,不依赖与其他第三方jar包,从而可以更好的减少依赖带来的问题。
  • 内核化:设计上,采用了微内核架构模式,内核小巧,实用,同时提供了一定的扩展性。从而可以更好地理解与应用本引擎。
  • 轻规范:并没有完全实现BPMN规范,也没有完全按照BPMN规范进行设计,而只是参考了该规范,且只实现以一小部分必须实现的元素。从而降低了学习成本,可以按照需求自由发挥。
  • 工具化:代码上,只是一个工具(UTIL),不是一个应用程序。从而你可以简单的运行它,扩展你自己的数据层、节点层,更加方便的集成到其他应用中去。

好,废话说完了,开始第一个迭代……

 

四、Hello ProcessEngine

按照国际惯例,第一个迭代用来实现 hello world 。

1、需求

作为一个流程管理员,我希望流程引擎可以运行如下图所示的流程,以便我能够配置流程来打印不同的字符串。

轻量级工作流引擎的设计与实现

 

2、分析

  • 第一个流程,可以打印Hello ProcessEngine,第二个流程可以打印ProcessEngine Hello,这两个流程的区别是只有顺序不同,蓝色的节点与红色的节点的本身功能没有发生变化
  • 蓝色的节点与红色的节点都是节点,它们的功能是不一样的,即:红色的节点打印Hello,蓝色的节点打印ProcessEngine
  • 开始与结束节点是两个特殊的节点,一个开始流程,一个结束流程
  • 节点与节点之间是通过线来连接的,一个节点执行完毕后,是通过箭头来确定下一个要执行的节点
  • 需要一种表示流程的方式,或是XML、或是JSON、或是其他,而不是图片

3、设计

(1)流程的表示

相较于JSON,XML的语义更丰富,可以表达更多的信息,因此这里使用XML来对流程进行表示,如下所示

<definitions>      <process id="process_1" name="hello">          <startEvent id="startEvent_1">              <outgoing>flow_1</outgoing>          </startEvent>          <sequenceFlow id="flow_1" sourceRef="startEvent_1" targetRef="printHello_1" />          <printHello id="printHello_1" name="hello">              <incoming>flow_1</incoming>              <outgoing>flow_2</outgoing>          </printHello>          <sequenceFlow id="flow_2" sourceRef="printHello_1" targetRef="printProcessEngine_1" />          <printProcessEngine id="printProcessEngine_1" name="processEngine">              <incoming>flow_2</incoming>              <outgoing>flow_3</outgoing>          </printProcessEngine>          <sequenceFlow id="flow_3" sourceRef="printProcessEngine_1" targetRef="endEvent_1"/>          <endEvent id="endEvent_1">              <incoming>flow_3</incoming>          </endEvent>      </process>  </definitions>

 

  • process表示一个流程
  • startEvent表示开始节点,endEvent表示结束节点
  • printHello表示打印hello节点,就是需求中的蓝色节点
  • processEngine表示打印processEngine节点,就是需求中的红色节点
  • sequenceFlow表示连线,从sourceRef开始,指向targetRef,例如:flow_3,表示一条从printProcessEngine_1到endEvent_1的连线。

(2)节点的表示

  • outgoing表示出边,即节点执行完毕后,应该从那个边出去。
  • incoming表示入边,即从哪个边进入到本节点。
  • 一个节点只有outgoing而没有incoming,如:startEvent,也可以 只有入边而没有出边,如:endEvent,也可以既有入边也有出边,如:printHello、processEngine。

(3)流程引擎的逻辑

基于上述XML,流程引擎的运行逻辑如下

  1. 找到开始节点(startEvent)
  2. 找到startEvent的outgoing边(sequenceFlow)
  3. 找到该边(sequenceFlow)指向的节点(targetRef)
  4. 执行节点自身的逻辑
  5. 找到该节点的outgoing边(sequenceFlow)
  6. 重复3-5,直到遇到结束节点(endEvent),流程结束

4、实现

首先要进行数据结构的设计,即:要把问题域中的信息映射到计算机中的数据。

可以看到,一个流程(PeProcess)由多个节点(PeNode)与边(PeEdge)组成,节点有出边(out)、入边(in),边有流入节点(from)、流出节点(to)。

具体的定义如下:

public class PeProcess {      public String id;      public PeNode start;        public PeProcess(String id, PeNode start) {          this.id = id;          this.start = start;      }  }    public class PeEdge {      private String id;      public PeNode from;      public PeNode to;        public PeEdge(String id) {          this.id = id;      }  }    public class PeNode {      private String id;        public String type;      public PeEdge in;      public PeEdge out;        public PeNode(String id) {          this.id=id;      }  }

PS : 为了表述主要思想,在代码上比较“奔放自由”,生产中不可直接复制粘贴!

接下来,构建流程图,代码如下:

public class XmlPeProcessBuilder {      private String xmlStr;      private final Map<String, PeNode> id2PeNode = new HashMap<>();      private final Map<String, PeEdge> id2PeEdge = new HashMap<>();        public XmlPeProcessBuilder(String xmlStr) {          this.xmlStr = xmlStr;      }        public PeProcess build() throws Exception {          //strToNode : 把一段xml转换为org.w3c.dom.Node          Node definations = XmlUtil.strToNode(xmlStr);          //childByName : 找到definations子节点中nodeName为process的那个Node          Node process = XmlUtil.childByName(definations, "process");          NodeList childNodes = process.getChildNodes();            for (int j = 0; j < childNodes.getLength(); j++) {              Node node = childNodes.item(j);              //#text node should be skip              if (node.getNodeType() == Node.TEXT_NODE) continue;                if ("sequenceFlow".equals(node.getNodeName()))                  buildPeEdge(node);              else                  buildPeNode(node);          }          Map.Entry<String, PeNode> startEventEntry = id2PeNode.entrySet().stream().filter(entry -> "startEvent".equals(entry.getValue().type)).findFirst().get();          return new PeProcess(startEventEntry.getKey(), startEventEntry.getValue());      }        private void buildPeEdge(Node node) {          //attributeValue : 找到node节点上属性为id的值          PeEdge peEdge = id2PeEdge.computeIfAbsent(XmlUtil.attributeValue(node, "id"), id -> new PeEdge(id));          peEdge.from = id2PeNode.computeIfAbsent(XmlUtil.attributeValue(node, "sourceRef"), id -> new PeNode(id));          peEdge.to = id2PeNode.computeIfAbsent(XmlUtil.attributeValue(node, "targetRef"), id -> new PeNode(id));      }        private void buildPeNode(Node node) {          PeNode peNode = id2PeNode.computeIfAbsent(XmlUtil.attributeValue(node, "id"), id -> new PeNode(id));          peNode.type = node.getNodeName();            Node inPeEdgeNode = XmlUtil.childByName(node, "incoming");          if (inPeEdgeNode != null)              //text : 得到inPeEdgeNode的nodeValue              peNode.in = id2PeEdge.computeIfAbsent(XmlUtil.text(inPeEdgeNode), id -> new PeEdge(id));            Node outPeEdgeNode = XmlUtil.childByName(node, "outgoing");          if (outPeEdgeNode != null)              peNode.out = id2PeEdge.computeIfAbsent(XmlUtil.text(outPeEdgeNode), id -> new PeEdge(id));      }  }

 

接下来,实现流程引擎主逻辑,代码如下:

public class ProcessEngine {      private String xmlStr;        public ProcessEngine(String xmlStr) {          this.xmlStr = xmlStr;      }        public void run() throws Exception {          PeProcess peProcess = new XmlPeProcessBuilder(xmlStr).build();            PeNode node = peProcess.start;          while (!node.type.equals("endEvent")) {              if ("printHello".equals(node.type))                  System.out.print("Hello ");              if ("printProcessEngine".equals(node.type))                  System.out.print("ProcessEngine ");                node = node.out.to;          }      }  }

就这?工作流引擎就这?同学们可千万不要这样简单理解啊,毕竟这还只是hello world而已,各种代码量就已经不少了。

另外,这里面还有很多可以改进的空间,比如异常控制、泛化、设计模式等,但毕竟只是一个hello world而已,其目的是方便同学理解,让同学入门。

那么,接下来呢,就要稍微贴近一些具体的实际应用场景了,我们继续第二个迭代。

五、简单审批

一般来讲工作流引擎属于底层技术,在它之上可以构建审批流、业务流、数据流等类型的应用,那么接下啦就以实际中的简单审批场景为例,继续深入工作流引擎的设计,好,我们开始。

1、需求

作为一个流程管理员,我希望流程引擎可以运行如下图所示的流程,以便我能够配置流程来实现简单的审批流。

轻量级工作流引擎的设计与实现

 

例如:小张提交了一个申请单,然后经过经理审批,审批结束后,不管通过还是不通过,都会经过第三步把结果发送给小张。

2、分析

  • 总体上来讲,这个流程还是线性顺序类的,基本上可以沿用上次迭代的部分设计
  • 审批节点的耗时可能会比较长,甚至会达到几天时间,工作流引擎主动式的调取下一个节点的逻辑并不适合此场景
  • 随着节点类型的增多,工作流引擎里写死的那部分节点类型自由逻辑也不合适
  • 审批时需要申请单信息、审批人,结果邮件通知还需要审批结果等信息,这些信息如何传递也是一个要考虑的问题

3、设计

  • 采用注册机制,把节点类型及其自有逻辑注册进工作流引擎,以便能够扩展更多节点,使得工作流引擎与节点解耦
  • 工作流引擎增加被动式驱动逻辑,使得能够通过外部来使工作流引擎执行下一个节点
  • 增加上下文语义,作为全局变量来使用,使得数据能够流经各个节点

4、实现

新的XML定义如下:

<definitions>      <process id="process_2" name="简单审批例子">          <startEvent id="startEvent_1">              <outgoing>flow_1</outgoing>          </startEvent>          <sequenceFlow id="flow_1" sourceRef="startEvent_1" targetRef="approvalApply_1" />          <approvalApply id="approvalApply_1" name="提交申请单">              <incoming>flow_1</incoming>              <outgoing>flow_2</outgoing>          </approvalApply>          <sequenceFlow id="flow_2" sourceRef="approvalApply_1" targetRef="approval_1" />          <approval id="approval_1" name="审批">              <incoming>flow_2</incoming>              <outgoing>flow_3</outgoing>          </approval>          <sequenceFlow id="flow_3" sourceRef="approval_1" targetRef="notify_1"/>          <notify id="notify_1" name="结果邮件通知">              <incoming>flow_3</incoming>              <outgoing>flow_4</outgoing>          </notify>          <sequenceFlow id="flow_4" sourceRef="notify_1" targetRef="endEvent_1"/>          <endEvent id="endEvent_1">              <incoming>flow_4</incoming>          </endEvent>      </process>  </definitions>

首先要有一个上下文对象类,用于传递变量的,定义如下:

public class PeContext {      private Map<String, Object> info = new ConcurrentHashMap<>();        public Object getValue(String key) {          return info.get(key);      }        public void putValue(String key, Object value) {          info.put(key, value);      }  }

每个节点的处理逻辑是不一样的,此处应该进行一定的抽象,为了强调流程中节点的作用是逻辑处理,引入了一种新的类型–算子(Operator),定义如下:

public interface IOperator {      //引擎可以据此来找到本算子      String getType();        //引擎调度本算子      void doTask(ProcessEngine processEngine, PeNode node, PeContext peContext);  }  

对于引擎来讲,当遇到一个节点时,需要调度之,但怎么调度呢?首先需要各个节点算子注册(registNodeProcessor())进来,这样才能找到要调度的那个算子。

其次,引擎怎么知道节点算子自有逻辑处理完了呢?一般来讲,引擎是不知道的,只能是由算子告诉引擎,所以引擎要提供一个功能(nodeFinished()),这个功能由算子调用。

最后,把算子任务的调度和引擎的驱动解耦开来,放入不同的线程中。

修改后的ProcessEngine代码如下:

public class ProcessEngine {      private String xmlStr;        //存储算子      private Map<String, IOperator> type2Operator = new ConcurrentHashMap<>();      private PeProcess peProcess = null;      private PeContext peContext = null;        //任务数据暂存      public final BlockingQueue<PeNode> arrayBlockingQueue = new LinkedBlockingQueue();      //任务调度线程      public final Thread dispatchThread = new Thread(() -> {          while (true) {              try {                  PeNode node = arrayBlockingQueue.take();                  type2Operator.get(node.type).doTask(this, node, peContext);              } catch (Exception e) {              }          }      });        public ProcessEngine(String xmlStr) {          this.xmlStr = xmlStr;      }        //算子注册到引擎中,便于引擎调用之      public void registNodeProcessor(IOperator operator) {          type2Operator.put(operator.getType(), operator);      }        public void start() throws Exception {          peProcess = new XmlPeProcessBuilder(xmlStr).build();          peContext = new PeContext();            dispatchThread.setDaemon(true);          dispatchThread.start();            executeNode(peProcess.start.out.to);      }        private void executeNode(PeNode node) {          if (!node.type.equals("endEvent"))              arrayBlockingQueue.add(node);          else              System.out.println("process finished!");      }        public void nodeFinished(String peNodeID) {          PeNode node = peProcess.peNodeWithID(peNodeID);          executeNode(node.out.to);      }  }

接下来,简单(简陋)实现本示例所需的三个算子,代码如下:

/**   * 提交申请单   */  public class OperatorOfApprovalApply implements IOperator {      @Override      public String getType() {          return "approvalApply";      }        @Override      public void doTask(ProcessEngine processEngine, PeNode node, PeContext peContext) {          peContext.putValue("form", "formInfo");          peContext.putValue("applicant", "小张");            processEngine.nodeFinished(node.id);      }  }    /**   * 审批   */  public class OperatorOfApproval implements IOperator {      @Override      public String getType() {          return "approval";      }        @Override      public void doTask(ProcessEngine processEngine, PeNode node, PeContext peContext) {          peContext.putValue("approver", "经理");          peContext.putValue("message", "审批通过");            processEngine.nodeFinished(node.id);      }  }    /**   * 结果邮件通知   */  public class OperatorOfNotify implements IOperator {      @Override      public String getType() {          return "notify";      }        @Override      public void doTask(ProcessEngine processEngine, PeNode node, PeContext peContext) {            System.out.println(String.format("%s 提交的申请单 %s 被 %s 审批,结果为 %s",                  peContext.getValue("applicant"),                  peContext.getValue("form"),                  peContext.getValue("approver"),                  peContext.getValue("message")));            processEngine.nodeFinished(node.id);      }  }

运行一下,看看结果如何,代码如下:

public class ProcessEngineTest {        @Test      public void testRun() throws Exception {          //读取文件内容到字符串          String modelStr = Tools.readResoucesFile("model/two/hello.xml");          ProcessEngine processEngine = new ProcessEngine(modelStr);            processEngine.registNodeProcessor(new OperatorOfApproval());          processEngine.registNodeProcessor(new OperatorOfApprovalApply());          processEngine.registNodeProcessor(new OperatorOfNotify());            processEngine.start();            Thread.sleep(1000 * 1);        }    }

 

小张 提交的申请单 formInfo 被 经理 审批,结果为 审批通过  process finished!

 

到此,轻量级工作流引擎的核心逻辑介绍的差不多了,然而,只支持顺序结构是太单薄的,我们知道,程序流程的三种基本结构为顺序、分支、循环,有了这三种结构,基本上就可以表示绝大多数流程逻辑。循环可以看做一种组合结构,即:循环可以由顺序与分支推导出来,我们已经实现了顺序,那么接下来只要实现分支即可,而分支有很多类型,如:二选一、N选一、N选M(1<=M<=N),其中N选一可以由二选一的组合推导出来,N选M也可以由二选一的组合推导出来,只是比较啰嗦,不那么直观,所以,我们只要实现二选一分支,即可满足绝大多数流程逻辑场景,好,第三个迭代开始。

 

六、一般审批

作为一个流程管理员,我希望流程引擎可以运行如下图所示的流程,以便我能够配置流程来实现一般的审批流。

轻量级工作流引擎的设计与实现

 

例如:小张提交了一个申请单,然后经过经理审批,审批结束后,如果通过,发邮件通知,不通过,则打回重写填写申请单,直到通过为止。

1、分析

  • 需要引入一种分支节点,可以进行简单的二选一流转
  • 节点的入边、出边不只一条
  • 需要一种逻辑表达式语义,可以配置分支节点

2、设计

  • 节点要支持多入边、多出边
  • 节点算子来决定从哪个出边出
  • 使用一种简单的规则引擎,支持简单的逻辑表达式的解析
  • 简单分支节点的XML定义

3、实现

新的XML定义如下:

<definitions>      <process id="process_2" name="简单审批例子">          <startEvent id="startEvent_1">              <outgoing>flow_1</outgoing>          </startEvent>          <sequenceFlow id="flow_1" sourceRef="startEvent_1" targetRef="approvalApply_1"/>          <approvalApply id="approvalApply_1" name="提交申请单">              <incoming>flow_1</incoming>              <incoming>flow_5</incoming>              <outgoing>flow_2</outgoing>          </approvalApply>          <sequenceFlow id="flow_2" sourceRef="approvalApply_1" targetRef="approval_1"/>          <approval id="approval_1" name="审批">              <incoming>flow_2</incoming>              <outgoing>flow_3</outgoing>          </approval>          <sequenceFlow id="flow_3" sourceRef="approval_1" targetRef="simpleGateway_1"/>          <simpleGateway id="simpleGateway_1" name="简单是非判断">              <trueOutGoing>flow_4</trueOutGoing>              <expr>approvalResult</expr>              <incoming>flow_3</incoming>              <outgoing>flow_4</outgoing>              <outgoing>flow_5</outgoing>          </simpleGateway>          <sequenceFlow id="flow_5" sourceRef="simpleGateway_1" targetRef="approvalApply_1"/>          <sequenceFlow id="flow_4" sourceRef="simpleGateway_1" targetRef="notify_1"/>          <notify id="notify_1" name="结果邮件通知">              <incoming>flow_4</incoming>              <outgoing>flow_6</outgoing>          </notify>          <sequenceFlow id="flow_6" sourceRef="notify_1" targetRef="endEvent_1"/>          <endEvent id="endEvent_1">              <incoming>flow_6</incoming>          </endEvent>      </process>  </definitions>

 

其中,加入了simpleGateway这个简单分支节点,用于表示简单的二选一分支,当expr中的表达式为真时,走trueOutGoing中的出边,否则走另一个出边。

节点支持多入边、多出边,修改后的PeNode如下:

public class PeNode {      public String id;        public String type;      public List<PeEdge> in = new ArrayList<>();      public List<PeEdge> out = new ArrayList<>();      public Node xmlNode;        public PeNode(String id) {          this.id = id;      }        public PeEdge onlyOneOut() {          return out.get(0);      }        public PeEdge outWithID(String nextPeEdgeID) {          return out.stream().filter(e -> e.id.equals(nextPeEdgeID)).findFirst().get();      }        public PeEdge outWithOutID(String nextPeEdgeID) {          return out.stream().filter(e -> !e.id.equals(nextPeEdgeID)).findFirst().get();      }    }

以前只有一个出边时,是由当前节点来决定下一节点的,现在多出边了,该由边来决定下一个节点是什么,修改后的流程引擎代码如下:

public class ProcessEngine {      private String xmlStr;        //存储算子      private Map<String, IOperator> type2Operator = new ConcurrentHashMap<>();      private PeProcess peProcess = null;      private PeContext peContext = null;        //任务数据暂存      public final BlockingQueue<PeNode> arrayBlockingQueue = new LinkedBlockingQueue();      //任务调度线程      public final Thread dispatchThread = new Thread(() -> {          while (true) {              try {                  PeNode node = arrayBlockingQueue.take();                  type2Operator.get(node.type).doTask(this, node, peContext);              } catch (Exception e) {                  e.printStackTrace();              }          }      });        public ProcessEngine(String xmlStr) {          this.xmlStr = xmlStr;      }        //算子注册到引擎中,便于引擎调用之      public void registNodeProcessor(IOperator operator) {          type2Operator.put(operator.getType(), operator);      }        public void start() throws Exception {          peProcess = new XmlPeProcessBuilder(xmlStr).build();          peContext = new PeContext();            dispatchThread.setDaemon(true);          dispatchThread.start();            executeNode(peProcess.start.onlyOneOut().to);      }        private void executeNode(PeNode node) {          if (!node.type.equals("endEvent"))              arrayBlockingQueue.add(node);          else              System.out.println("process finished!");      }        public void nodeFinished(PeEdge nextPeEdgeID) {          executeNode(nextPeEdgeID.to);      }  }

新加入的simpleGateway节点算子如下:

/**   * 简单是非判断   */  public class OperatorOfSimpleGateway implements IOperator {      @Override      public String getType() {          return "simpleGateway";      }        @Override      public void doTask(ProcessEngine processEngine, PeNode node, PeContext peContext) {          ScriptEngineManager manager = new ScriptEngineManager();          ScriptEngine engine = manager.getEngineByName("js");          engine.put("approvalResult", peContext.getValue("approvalResult"));            String expression = XmlUtil.childTextByName(node.xmlNode, "expr");          String trueOutGoingEdgeID = XmlUtil.childTextByName(node.xmlNode, "trueOutGoing");            PeEdge outPeEdge = null;          try {              outPeEdge = (Boolean) engine.eval(expression) ?                      node.outWithID(trueOutGoingEdgeID) : node.outWithOutID(trueOutGoingEdgeID);          } catch (ScriptException e) {              e.printStackTrace();          }            processEngine.nodeFinished(outPeEdge);      }  }

其中简单使用了js脚本作为表达式,当然其中的弊端这里就不展开了。

为了方便同学们CC+CV,其他发生相应变化的代码如下:

/**   * 审批   */  public class OperatorOfApproval implements IOperator {      @Override      public String getType() {          return "approval";      }        @Override      public void doTask(ProcessEngine processEngine, PeNode node, PeContext peContext) {          peContext.putValue("approver", "经理");            Integer price = (Integer) peContext.getValue("price");          //价格<=200审批才通过,即:approvalResult=true          boolean approvalResult = price <= 200;          peContext.putValue("approvalResult", approvalResult);            System.out.println("approvalResult :" + approvalResult + ",price : " + price);            processEngine.nodeFinished(node.onlyOneOut());      }  }    /**   * 提交申请单   */  public class OperatorOfApprovalApply implements IOperator {        public static int price = 500;        @Override      public String getType() {          return "approvalApply";      }        @Override      public void doTask(ProcessEngine processEngine, PeNode node, PeContext peContext) {          //price每次减100          peContext.putValue("price", price -= 100);          peContext.putValue("applicant", "小张");            processEngine.nodeFinished(node.onlyOneOut());      }  }      /**   * 结果邮件通知   */  public class OperatorOfNotify implements IOperator {      @Override      public String getType() {          return "notify";      }        @Override      public void doTask(ProcessEngine processEngine, PeNode node, PeContext peContext) {            System.out.println(String.format("%s 提交的申请单 %s 被 %s 审批,结果为 %s",                  peContext.getValue("applicant"),                  peContext.getValue("price"),                  peContext.getValue("approver"),                  peContext.getValue("approvalResult")));            processEngine.nodeFinished(node.onlyOneOut());      }  }      public class XmlPeProcessBuilder {      private String xmlStr;      private final Map<String, PeNode> id2PeNode = new HashMap<>();      private final Map<String, PeEdge> id2PeEdge = new HashMap<>();        public XmlPeProcessBuilder(String xmlStr) {          this.xmlStr = xmlStr;      }        public PeProcess build() throws Exception {          //strToNode : 把一段xml转换为org.w3c.dom.Node          Node definations = XmlUtil.strToNode(xmlStr);          //childByName : 找到definations子节点中nodeName为process的那个Node          Node process = XmlUtil.childByName(definations, "process");          NodeList childNodes = process.getChildNodes();            for (int j = 0; j < childNodes.getLength(); j++) {              Node node = childNodes.item(j);              //#text node should be skip              if (node.getNodeType() == Node.TEXT_NODE) continue;                if ("sequenceFlow".equals(node.getNodeName()))                  buildPeEdge(node);              else                  buildPeNode(node);          }          Map.Entry<String, PeNode> startEventEntry = id2PeNode.entrySet().stream().filter(entry -> "startEvent".equals(entry.getValue().type)).findFirst().get();          return new PeProcess(startEventEntry.getKey(), startEventEntry.getValue());      }        private void buildPeEdge(Node node) {          //attributeValue : 找到node节点上属性为id的值          PeEdge peEdge = id2PeEdge.computeIfAbsent(XmlUtil.attributeValue(node, "id"), id -> new PeEdge(id));          peEdge.from = id2PeNode.computeIfAbsent(XmlUtil.attributeValue(node, "sourceRef"), id -> new PeNode(id));          peEdge.to = id2PeNode.computeIfAbsent(XmlUtil.attributeValue(node, "targetRef"), id -> new PeNode(id));      }        private void buildPeNode(Node node) {          PeNode peNode = id2PeNode.computeIfAbsent(XmlUtil.attributeValue(node, "id"), id -> new PeNode(id));          peNode.type = node.getNodeName();          peNode.xmlNode = node;            List<Node> inPeEdgeNodes = XmlUtil.childsByName(node, "incoming");          inPeEdgeNodes.stream().forEach(n -> peNode.in.add(id2PeEdge.computeIfAbsent(XmlUtil.text(n), id -> new PeEdge(id))));            List<Node> outPeEdgeNodes = XmlUtil.childsByName(node, "outgoing");          outPeEdgeNodes.stream().forEach(n -> peNode.out.add(id2PeEdge.computeIfAbsent(XmlUtil.text(n), id -> new PeEdge(id))));      }  }

运行一下,看看结果如何,代码如下:

public class ProcessEngineTest {        @Test      public void testRun() throws Exception {          //读取文件内容到字符串          String modelStr = Tools.readResoucesFile("model/third/hello.xml");          ProcessEngine processEngine = new ProcessEngine(modelStr);            processEngine.registNodeProcessor(new OperatorOfApproval());          processEngine.registNodeProcessor(new OperatorOfApprovalApply());          processEngine.registNodeProcessor(new OperatorOfNotify());          processEngine.registNodeProcessor(new OperatorOfSimpleGateway());            processEngine.start();            Thread.sleep(1000 * 1);      }    }

 

approvalResult :false,price : 400  approvalResult :false,price : 300  approvalResult :true,price : 200  小张 提交的申请单 200  经理 审批,结果为 true  process finished!

 

至此,本需求实现完毕,除了直接实现了分支语义外,我们看到,这里还间接实现了循环语义。

作为一个轻量级的工作流引擎,到此就基本讲完了,接下来,我们做一下总结与展望。

 

七、总结与展望

经过以上三个迭代,我们可以得到一个相对稳定的工作流引擎的结构,如下图所示:

轻量级工作流引擎的设计与实现

 

通过此图我们可知,这里有一个相对稳定的引擎层,同时为了提供扩展性,提供了一个节点算子层,所有的节点算子的新增都在此处中。

此外,进行了一定程度的控制反转,即:由算子决定下一步走哪里,而不是引擎。这样,极大地提高了引擎的灵活性,更好的进行了封装。

最后,使用了上下文,提供了一种全局变量的机制,便于节点之间的数据流动。

当然,以上的三个迭代距离实际的线上应用场景相距甚远,还需实现与展望以下几点才可,如下:

  • 一些异常情况的考虑与设计
  • 应把节点抽象成一个函数,要有入参、出参,数据类型等
  • 关键的地方加入埋点,用以控制引擎或吐出事件
  • 图的语义合法性检查,xsd、自定义检查技术等
  • 图的dag算法检测
  • 流程的流程历史记录,及回滚到任意节点
  • 流程图的动态修改,即:可以在流程开始后,对流程图进行修改
  • 并发修改情况下的考虑
  • 效率上的考虑
  • 防止重启后流转信息丢失,需要持久化机制的加入
  • 流程的取消、重置、变量传入等
  • 更合适的规则引擎及多种规则引擎的实现、配置
  • 前端的画布、前后端流程数据结构定义及转换

 

作者:刘洋

本文来自网络,不代表技术学习分享_CKX技术立场,转载请注明出处。

作者: CKX技术

上一篇
下一篇
广告位

发表回复

返回顶部