标签：Run 流程图 队列 流程 节点 可视化 java label id

前言

首先，感谢大家对上一篇文章[业务可视化-让你的流程图"Run"起来(2.问题与改进)]的支持。

分享一下近期我对这个项目的一些改进。

1. 增加了分支选择工程，可以根据节点的运行结果决定执行哪一个节点。

2. 增加了分布式运行节点功能，可以将流程节点部署到任意服务器，通过队列来调度节点，也就是说节点的运行将不在局限于Java语言。

1. 如何让流程图“Run”起来

首先我们回顾一下前两篇文章的知识，如何让流程图“Run”起来：

工程目录[ html/network.html ]里，提供了一个图形化界面的流程设计器，可以通过拖拽的方式设计流程并生成Json文件。

反之，也可以将Json文件转化为流程图并进行编辑。

 

1.1 创建流程

创建流程过程如下图所示：

节点和边的ID自动生成，可以自己定义节点名和每个边对应的节点返回值。

点击[ update json ]后，即可生成/更新流程图对应的Json文件。

 

1.2 节点与被执行的Java方法绑定

我们需要写一个Java类，继承自FlowRunner。

然后在里面写每个节点对应的方法，用@Node注释来实现与流程中节点的绑定。

同时将1.1中生成的Json文件放到和Java类相同的目录下。

TestFlow1.java

public class TestFlow1 extends FlowRunner {

	@Node(label = "a")
	public int process_a() {
		System.out.println("processing a");
		return 1;
	}

	@Node(label = "b")
	public void process_b() {
		System.out.println("processing b");
	}

	@Node(label = "c")
	public void process_c() {
		System.out.println("processing c");
	}

	@Node(label = "d")
	public void process_d() {
		System.out.println("processing d");
	}
}

TestFlow1.json

{
	"flowId": "your flow id",
	"nodes": [
		{
			"id": "e21eb7b6-2f23-4264-a50f-e42321dd295b",
			"label": "a",
			"readyCheck": 0
		},
		{
			"id": "f2a76819-b6a8-49db-af25-fab8274550f3",
			"label": "b",
			"readyCheck": 0
		},
		{
			"id": "73f8bd68-8454-4b02-9098-c0c7bb6ffdb2",
			"label": "c",
			"readyCheck": 0
		},
		{
			"id": "3553d1f7-e4c3-4e4b-a9ef-80b94ebbb8af",
			"label": "d",
			"readyCheck": 1
		}
	],
	"edges": [
		{
			"id": "36bdc526-f6ae-45de-9bb7-34c293b34006",
			"from": "e21eb7b6-2f23-4264-a50f-e42321dd295b",
			"to": "f2a76819-b6a8-49db-af25-fab8274550f3",
			"condition": "1",
			"arrows": "to"
		},
		{
			"id": "652b871d-338d-45f5-91a9-3a488ed9b6f4",
			"from": "e21eb7b6-2f23-4264-a50f-e42321dd295b",
			"to": "73f8bd68-8454-4b02-9098-c0c7bb6ffdb2",
			"condition": "2",
			"arrows": "to"
		},
		{
			"id": "2691b6fe-ede9-4d1c-8b49-82d2a4ef014a",
			"from": "f2a76819-b6a8-49db-af25-fab8274550f3",
			"to": "3553d1f7-e4c3-4e4b-a9ef-80b94ebbb8af",
			"arrows": "to"
		},
		{
			"id": "d8026555-7609-4d27-8689-fd3dbcfe11d7",
			"from": "73f8bd68-8454-4b02-9098-c0c7bb6ffdb2",
			"to": "3553d1f7-e4c3-4e4b-a9ef-80b94ebbb8af",
			"arrows": "to"
		}
	]
}

Test1.java

public class Test1 {

	public static void main(String[] args) {
		
		TestFlow1 testFlow = new TestFlow1();
		testFlow.startFlow(true);
		
	}
}

 

1.3 启动流程

调用1.2中写好的Java类的startFlow方法，即可启动流程。

同步启动

TestFlow1 testFlow = new TestFlow1();
testFlow.startFlow(true);

异步启动

TestFlow1 testFlow = new TestFlow1();
testFlow.startFlow(false);

1.4 关闭流程执行器

流程执行器会在第一个流程启动的时候自动启动，在整个系统关闭的时候，我们需要将流程执行器关闭，如下。

FlowStarter.shutdown();

1.5 流程执行结果确认

在流程执行完毕后，日志会输出执行结果的json文件，我们可以将这个文件粘贴到1.1介绍的工具里，生成图形化的执行结果来确认节点的运行状况。

运行成功日志

Ready queue thread started.
Complete queue thread started.
json:
{"flowId":"123","nodes":[{"id":"1","label":"a"},{"id":"2","label":"b"},{"id":"0b5ba9df-b6c7-4752-94e2-debb6104015c","label":"c"},{"id":"29bc32c7-acd8-4893-9410-e9895da38b2e","label":"d"}],"edges":[{"id":"1","from":"1","to":"2","arrows":"to"},{"id":"078ffa82-5eff-4d33-974b-53890f2c9a18","from":"1","to":"0b5ba9df-b6c7-4752-94e2-debb6104015c","arrows":"to"},{"id":"90663193-7077-4aca-9011-55bc8745403f","from":"2","to":"29bc32c7-acd8-4893-9410-e9895da38b2e","arrows":"to"},{"id":"a6882e25-c07a-4abd-907e-e269d4eda0ec","from":"0b5ba9df-b6c7-4752-94e2-debb6104015c","to":"29bc32c7-acd8-4893-9410-e9895da38b2e","arrows":"to"}]}
execute:1
node name:a
processing a
execute:2
node name:b
processing b
execute:0b5ba9df-b6c7-4752-94e2-debb6104015c
node name:c
processing c
execute:29bc32c7-acd8-4893-9410-e9895da38b2e
node name:d
processing d
Complete success.
json:
{"nodes":[{"id": "1","label": "a" ,"color": "#36AE7C"},{"id": "2","label": "b" ,"color": "#36AE7C"},{"id": "0b5ba9df-b6c7-4752-94e2-debb6104015c","label": "c" ,"color": "#36AE7C"},{"id": "29bc32c7-acd8-4893-9410-e9895da38b2e","label": "d" ,"color": "#36AE7C"}],"edges":[{"id": "1","from": "1","to": "2","arrows": "to"},{"id": "078ffa82-5eff-4d33-974b-53890f2c9a18","from": "1","to": "0b5ba9df-b6c7-4752-94e2-debb6104015c","arrows": "to"},{"id": "90663193-7077-4aca-9011-55bc8745403f","from": "2","to": "29bc32c7-acd8-4893-9410-e9895da38b2e","arrows": "to"},{"id": "a6882e25-c07a-4abd-907e-e269d4eda0ec","from": "0b5ba9df-b6c7-4752-94e2-debb6104015c","to": "29bc32c7-acd8-4893-9410-e9895da38b2e","arrows": "to"}]}

流程执行结束后，会输出执行结果和运行后的流程图状态。
可以直接将json贴到下面的位置，查看看结果（绿色表示正常结束，红色表示异常结束，白色表示等待执行）。

运行失败日志

Ready queue thread started.
Complete queue thread started.
json:
{"flowId":"123","nodes":[{"id":"1","label":"a"},{"id":"2","label":"b"},{"id":"0b5ba9df-b6c7-4752-94e2-debb6104015c","label":"c"},{"id":"29bc32c7-acd8-4893-9410-e9895da38b2e","label":"d"}],"edges":[{"id":"1","from":"1","to":"2","arrows":"to"},{"id":"078ffa82-5eff-4d33-974b-53890f2c9a18","from":"1","to":"0b5ba9df-b6c7-4752-94e2-debb6104015c","arrows":"to"},{"id":"90663193-7077-4aca-9011-55bc8745403f","from":"2","to":"29bc32c7-acd8-4893-9410-e9895da38b2e","arrows":"to"},{"id":"a6882e25-c07a-4abd-907e-e269d4eda0ec","from":"0b5ba9df-b6c7-4752-94e2-debb6104015c","to":"29bc32c7-acd8-4893-9410-e9895da38b2e","arrows":"to"}]}
execute:1
node name:a
processing a
execute:2
node name:b
processing b
execute:0b5ba9df-b6c7-4752-94e2-debb6104015c
node name:c
processing c
java.lang.reflect.InvocationTargetException
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at io.github.nobuglady.network.fw.FlowRunner.execute(FlowRunner.java:49)
	at io.github.nobuglady.network.fw.executor.NodeRunner.run(NodeRunner.java:93)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: java.lang.RuntimeException: test
	at io.github.nobuglady.network.MyFlow1.process_b(MyFlow1.java:16)
	... 11 more
java.lang.reflect.InvocationTargetException
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.base/java.lang.reflect.Method.invoke(Method.java:566)
	at io.github.nobuglady.network.fw.FlowRunner.execute(FlowRunner.java:49)
	at io.github.nobuglady.network.fw.executor.NodeRunner.run(NodeRunner.java:93)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: java.lang.RuntimeException: test
	at io.github.nobuglady.network.MyFlow1.process_b(MyFlow1.java:16)
	... 11 more
Complete error.
json:
{"nodes":[{"id": "1","label": "a" ,"color": "#36AE7C"},{"id": "2","label": "b" ,"color": "#EB5353"},{"id": "0b5ba9df-b6c7-4752-94e2-debb6104015c","label": "c" ,"color": "#36AE7C"},{"id": "29bc32c7-acd8-4893-9410-e9895da38b2e","label": "d" ,"color": "#E8F9FD"}],"edges":[{"id": "1","from": "1","to": "2","arrows": "to"},{"id": "078ffa82-5eff-4d33-974b-53890f2c9a18","from": "1","to": "0b5ba9df-b6c7-4752-94e2-debb6104015c","arrows": "to"},{"id": "90663193-7077-4aca-9011-55bc8745403f","from": "2","to": "29bc32c7-acd8-4893-9410-e9895da38b2e","arrows": "to"},{"id": "a6882e25-c07a-4abd-907e-e269d4eda0ec","from": "0b5ba9df-b6c7-4752-94e2-debb6104015c","to": "29bc32c7-acd8-4893-9410-e9895da38b2e","arrows": "to"}]}

流程执行结束后，会输出执行结果和运行后的流程图状态。
可以直接将json贴到下面的位置，查看看结果（绿色表示正常结束，红色表示异常结束，白色表示等待执行）。

2. 分支选择的改进

2.1 分支选择定义

我们可以在图形界面中定义每条边的值，流程运行时，对节点的返回值与后续边的值进行比对，比对结果一直则执行该条边对应的后续节点。

2.2 节点启动条件定义

对于多条边Join到一个节点的情况，我们需要定义该节点启动的条件，如下：

1. 指向该节点的任意一条边通过检查，则启动该节点

2. 指想该节点的所有边都通过检车后，启动该节点

上图表示节点b和节点c 任意一个节点完成后，执行节点d

 

2.3 节点返回值绑定

节点的返回值与Java的方法返回值自动绑定，流程执行后，

对于有返回值的方法，则会调用该返回值的toString方法作为该节点的返回值。

对于无返回值的方法，则默认空文字列为返回值。

比如，返回int值，则用返回的int值与后续边的条件做对比。

@Node(label = "a")
	public int process_a() {
		System.out.println("processing a");
		return 1;
	}

返回String值，则用返回的String值与后续边的条件做对比。

@Node(label = "a")
	public String process_a() {
		System.out.println("processing a");
		return "1";
	}

返回自定义Object等，则用返回的Ojbect值的toString()方法生成的字符串与后续边的条件做对比。

	@Node(label = "a")
	public MyObj process_a() {
		System.out.println("processing a");
		return new MyObj();
	}

2.4 节点间参数传递

目前还没有对节点间参数传递做特别的处理，

可以通过类变量等方式进行节点间参数的传递。

3. 分布式运行的改进

3.1 系统结构

把单体的工程改进成分布式的工程，首先要明确系统结构和改进点。

目前的系统结构如下图所示(黄色部分可以配置成分布式运行)

系统通过两个队列来进行节点间控制信息的流转。

1. 待启动队列

2. 完成队列

3.1.1 待启动队列

生产者：流程管理器（FlowManager），流程启动后，流程管理器将初始节点放入[待启动队列]中，等待消费。

消费者：流程执行器（NodeExecutor），流程执行器监听[待启动队列]，得到消息后，根据节点信息运行该节点，运行完成后，将节点的运行结果放入[完成队列]中，等待消费。

3.1.2 完成队列

生产者：流程执行器（NodeExecutor），流程执行器监听[待启动队列]，得到消息后，根据节点信息运行该节点，运行完成后，将节点的运行结果放入[完成队列]中，等待消费。

消费者：流程管理器（FlowManager），流程管理器监听[完成队列]，得到消息后，根据完成节点的信息，更新流程图，然后将后续待启动的节点放入[待启动队列]中，等待消费。

3.2 分布式系统改进

基于3.1介绍的系统结构，可以明显的发现队列是单机系统改进为分布式系统的改进点。

所以，把队列变成可配置的队列后，系统将可以通过配置文件选择单机部署，或者分布式部署。

配置文件如下

node.executor.remote=false
queue.ready.manager=io.github.nobuglady.network.fw.queue.ready.ReadyQueueManager
queue.complete.manager=io.github.nobuglady.network.fw.queue.complete.CompleteQueueManager
node.executor=io.github.nobuglady.network.fw.executor.NodePool

 

node.executor

对节点的执行器的配置，系统默认提供了本地的执行器，可以通过Annotation对节点绑定的方法进行调用。

您可以配置自己的节点执行器，需要实现接口INodeExecutor

  onNodeReady：节点准备运行的时候，会调用这个方法

  这个方法里需要写节点执行的具体方法，并且在节点执行完毕后，将节点运行结果放入[完成队列]。

  shutdown：系统关闭的时候，会调用这个方法

node.executor.remote

false:本地执行

true:远程执行

本地执行时，会调用node.executor中配置的执行器，来执行节点的运行。

远程执行时，则系统不会启动 流程执行器（NodeExecutor）。也就是不会消费[待启动队列]中的消息。

远程执行时，目标系统监听[待启动队列]的消息，得到消息后，根据节点信息运行该节点，运行完成后，将节点的运行结果放入[完成队列]中，等待消费。

所以[待启动队列]和[完成队列]必须配置成远程系统可以访问的队列。

 

queue.ready.manager

待启动队列管理器

需要提供队列的消费和生产的方法

注：配置成远程执行节点时，系统不会调用此队列的消费方法。（由远程系统消费此队列信息）

 

queue.complete.manager

完成队列管理器

需要提供队列的消费和生产的方法

注：配置成远程执行节点时，系统不会调用此队列的生产方法。（由远程系统生产此队列信息）

 

5. 本地运行和分布式运行配置例

下面介绍以RabbiMQ作为远程队列，进行分布式调用的配置，选择其他的远程队列可以酌情修改。

工程里的test1-6分别对应如下6种启动方式，Test1-6.java为启动类。

每次启动之前，需要修改ladybugflow1-6.properties为ladybugflow.properties

1.  默认配置：通过流程类启动流程

启动代码

TestFlow1 testFlow = new TestFlow1();
testFlow.startFlow(true);
FlowStarter.shutdown();

2. 本地节点：自定义【待启动队列】和【完成队列】

启动代码

TestFlow2 testFlow = new TestFlow2();
testFlow.startFlow(true);
FlowStarter.shutdown();

3. 本地节点：自定义【节点执行器】

启动代码

TestFlow3 testFlow = new TestFlow3();
testFlow.startFlow(true);
FlowStarter.shutdown();

4. 远程节点：通过流程类启动流程

启动代码

TestFlow4 testFlow = new TestFlow4();
testFlow.startFlow(true);
FlowStarter.shutdown();

5. 默认配置：通过指定Json文件来启动流程

启动代码

FlowRunner flowRunner = new FlowRunner(new TestFlow5());
flowRunner.startFlowFromJson("io/github/nobuglady/network/demo/test5/TestFlow5.json", true);
FlowStarter.shutdown();

6. 远程节点：通过指定Json文件来启动流程

启动代码

FlowRunner flowRunner = new FlowRunner();
flowRunner.startFlowFromJson("io/github/nobuglady/network/demo/test5/TestFlow5.json", true);
FlowStarter.shutdown();

感谢您看文章读到这里。

最后

源码：https://github.com/nobuglady/ladybugflow

运行例源码：https://github.com/nobuglady/ladybugflow-demo

运行例源码（远程节点）：https://github.com/nobuglady/ladybugflow-demo-remote-app

标签：Run,流程图,队列,流程,节点,可视化,java,label,id
来源： https://www.cnblogs.com/nobuglady/p/16474433.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。