OBE 的分析报告（纯JAVA版本的OBE）

OBE 有2个版本：
         一个是纯java的，另外一个是在此基础上改进的基于J2EE的版本，目前能google到的基本是J2EE版本。但J2EE版本过于庞大，容易陷入细节；而纯java的版本就结构很清晰，容易理解，是我当年（2004年）开始研究工作流时候的第一个版本。

         这个版本的分析如下：在文章的后面是for eclipse工程的workflow core代码，对于希望研究工作流引擎实现的有志同仁来说，这个版本是最容易上手的。

        OBE(open business engine)是开放源码的, 是基于java 开发的,面向WFMC 工作流规范的工作流软件. OBE 的源代码主要包括四个部分:OBE core, OBEengine, OBE test, OBE Designer。


           本文先对各个部分进行分析，然后综合分析。工作流引擎的大致处理过程，流程中活动执行相关顺序的控制。

•  OBE core 是标准的OBE API, 包含了针对xpdl 描述的工作流流程定义而设定的一系列的基本元素类，分别用于表述工作流的各个基本组成元素信息，比如activity，participant，transition 等。同时，OBE core 还包括了用于OBE 中的日志管理。
•  OBE Engine 包括了引擎的具体实现部分。
•  OBE Test 是一个简单的应用测试程序。
•  OBE Designer 是一个图形化的工作流建模工具。

下面针对于每个部分进行单独的结构组成以及功能分析。
1. OBE core
OBE core 的结构基本上是针对于工作流定义的结构来组织的, 由以下的package 组成:
1. org.obe 定义了描述用倒的抽象基础类
2. org.obe.activity 用于描述工作流定义中的活动描述
3. org.obe.applition 用于描述工作流定义中的应用描述
4. org.obe.condition 用于描述工作流定义中的条件
5. org.obe.data 用于描述工作流定义中的数据,包括参数等
6. org.obe.log 描述工作流日志 org.obe.log.provider
7. org.obe.parser 将xpdl 文件解析,形成package 类
8. org.obe.participant 用于描述工作流定义中的参与者
9. org.obe.serializer 将obe 中的工作流模型(package 类)序列化,写入输出流
10. org.obe.transition 用于描述工作流定义中的转移
11. org.obe.util 用于工作流中的时间信息的描述
12. org.obe.workflow 用于工作流的整体的描述

下面针对每一个package 进行分析：
a.  org.obe
包括以下几个部分：
1. AbstractWFElement 对于所有流程的基本元的抽象接口
2. ConformanceClass 对于阻塞状态的描述:FULL_BLOCKED,LOOP_BLOCKED
3. ExternalPackage  定义了包头的扩展部分
4. OBEException   定义了各种异常的接口，以利于后面的调用
5. Package 描述了xpdl 中定义的package
6. PackageHeader 描述了一个package 的标准的头部
7. PublicationStatus 描述流程定义中的三种状态:UNDER_REVISION,RELEASED, UNDER_TEST
8. RedefinableHeader 描述可用于实体的头部信息
9. WFElement  定义了工作流的元素共有的基本属性。有ID,Name, Description, ExtendedAttributes.


b.  org.obe.activity:
其中定义了工作流定义中的activity 的参数,包括活动的类型(自动活动,路由活动,人工活动),活动执行的类型(同步异步).自动活动的实现分类（LoopType， SubFlowType，ToolType)。对于Loop, 支持While-Loop 和repeat-Loop 两种。对于Tool 的类型, 支持application 和procedure 两种，application 指的是任何在自己的环境中执行的应用，而procedure 指的是在工作流的环境中执行的程序。

1. Activity   获取和设置工作流活动的各项属性，如finishMode,icon,limit等等
2. AutomationMode 用于描述活动是人工活动还是自动活动;
3. ExecutionType 用于描述执行的方式是同步还是异步;
4. Implementation 定义一个接口
5. ImplementationType 用于描述活动的实现类型,
6. Loop 描述loop,包括loop 的条件以及loop 的种类
7. LoopType 定义了loop 的两种类型
8. NoImplementation 针对于活动实现是空的情况
9. Route 针对于路由活动，这里只定义了一个类，没有属性和方法，客队之扩展。
10. SubFlow 用于描述在当前工作流处理环境中执行的另一个流程处理
11. Tool 描述实现的类型以及相应的参数
12. 由于core 的设计的出发点就是描述流程,所以所有的类中提供的方法主要是get 和set 相关的属性, 用于设置和读取相应的描述.
13. ToolSet  设置工作流执行的工具 如application ,procedure.
14. ToolType  定义了工具的类型，即application ,procedure.


c.  org.obe.applition
1. Application: 定义了流程中可以访问的应用的基本接口,(interface)
2. AbstractApplication: 抽象实现
3. BasicApplication: 具体的实现
继承关系:
AbstractWFElement=> Application=> AbstractApplication=> BasicApplication


d.  org.obe.condition
1. Abstract Condition 抽象实现类
2. Abstract Xpression 抽象实现类
3. BasicCondition  来自于 Abstract Condition，但暂没内容
4. BasicXpression  来自于 Abstract Xpression，但暂没内容
5. Condition(interface):描述状态的接口
6. ConditionType:描述了条件的类型condition 和otherwise
7. Xpression(interface): 描述状态的接口

e.  org.obe.data
描述了流程中用到的数据的表示:
1. ActualParameter:定义了从流程相关的数据传到一个活动应用中的参数
2. ArrayType: 描述了数组类型
3. BasicType: 包含了基本的数据类型:string,reference, float, int, 和类型datetime,
4. DataField:描述了在工作流执行的时候存放工作流相关数据的数据域
5. DataType:存放Type
6. DeclaredType: 对新声明类型的描述
7. EnumerationType: 枚举类型
8. EnumerationValue: 枚举值
9. FormalParameter:形式参数的描述,包括类型,描述,id,index,名称
10. ListType 
11. ParameterMode:参数模式: IN,INOUT,OUT
12. PlainType: 普通类型,可以设置为boolean,unit 和performer
13. RecordType: 记录类型
14. Type(interface): 类型定义的基本接口,
15. TypeDeclaration:类型声明
16. UnionType


f.  org.obe.log
1. Logger(interface):标准日志接口,包含fatal,error,warn,info 和debug 方法
2. LogManager: 用于管理log, 准确的说是来设置LogProvider,其中包含一张HashMap,记录了provider 的相关记录,例如:providerMap.put("sysout", "org.obe.log.provider.SysoutLogProvider");;
3. LogProvider(interface):对logger Object 提供者的接口,在org.obe.logger.provider 中实现不同的provider;


org. obe.logger.provider
1. 实现了provider 接口,针对不同的方式生成不同的provider,已有的provider 包括：NullLogProvider 和SysoutLogProvider


g.  org.obe.parser
对xpdl 文件的解析:
1. XPDLNames(interface):定义了xpdl 中的tag 的名字,将他们设为静态常量
2. XPDLParser(interface):定义了xpdl 分析器的接口(将输入流处理后,输出为Package 类型)
3. XPDLParserException:
4. ElementRequiredException

org.obe.parser.dom4j
利用dom4j 来实现xpdlparser
Dom4jXPDLParser 具体实现类,实现对于xpdl 的解析,其中用了dom4j
Util 提供了从Element 中提取相应name 值的方法


h.  org.obe.participant
WFElement=>Participant=>abstractParticipant=>BasicParticipant
描述了参与者的定义;
ParticipantType: 定义了参与者的类型, 包括HUMAN,
ORGANIZATIONAL_UNIT, SYSTEM, ROLE, RESOURCE, RESOURCE_SET


i.  org.obe.serializer
与parser 相反,将一个package 序列化为一个xpdl 文件
XPDLSerializer (interface):定义了Serializer 一个package 的接口(将package 输出为输出流)


org.obe. serializer.dom4j
利用dom4j 来实现XPDLSerializer
Dom4jXPDLSerializer 具体实现类,实现对于package 的序列化,其中用了dom4j
Util 提供了向Element 中添加name 以及相应值的方法


j.  org.obe.transition
WFElement=> transition=> abstract transition=>BasicTransition
描述了活动转移,包括from,to,condition, TransitionLoopType
InlineBlock: 对于阻塞的描述,包括起点,终点
Join: 用于构建join
JoinType: 对于join 的类型的描述
Split:
splitType
TransitionLoopType:对于转移中的loop 属性:no-loop,from-loop,to-loop
TransitionReference 包含transition 的id 号
TransitionRestriction 转移约束包括join,split 和InlineBlock


k.  org.obe.util
描述了流程定义中的时间设定部分
Duration:对于持续时间的描述
DurationUnit:持续时间计量单位设定,包括年月日小时分钟秒;
InstantiationType:实例化类型包括ONCE 和MULTIPLE 两种
*SimulationInformation:时间信息描述,
包括了timeEstimation 和instantiationType
TimeEstimation 时间计划包括等待时间,工作时间和持续时间


l.  org.obe.workflow
1. WorkflowProcess 描述工作流接口
2. AbstractWorkflowProcess 抽象实现类，主要由他来实现。
3. BasicWorkflowProcess 基本实现类
4. ProcessHeader 取得过程头部信息
5. AbstractWFElement=> WorkflowProcess=> AbstractWorkflowProcess => BasicWorkflowProcess


2. OBE engine
Obe engine 包含了实现工作流流程控制的接口,主要由以下的package 来组成:

A  org.obe.action
描述了在流程执行中的一些基本的动作，例如给相关的参与者发送通知等，其中Action 是基本的动作接口定义，NotifyResponsibleAction 是一个Action 的实现类，实现了当一个活动超出了时间限制或者有错误发生的时候,向相关的参与者发送通知。

B  org.obe.engine
本部分是obe 工作流引擎控制部分的核心，主要包含对于活动列表的描述，对执行活动的控制部分以及对于整个工作流的整体描述和实例化部分。

主要控制部分：
1. WorkflowEngine 是对外提供应用的接口，功能包括初始化Repository- Manager和package，同步或异步执行。 它是通过创建一个WorkflowProcessInstance 来进行工作流的处理。
2. WorkflowProcessInstance 是一个workflowEngine 的实例，描述了一个流程执行过程中的状态以及控制整个处理过程。
3. WorkflowRunner 提供了流程的执行逻辑，解决了流程中活动的转移，实现了流程的运行控制。

对于workitem 的描述部分:
1. workItem 包含了对于一个活动的描述,包括workflowContext，activity和状态。
2. workItemState: 描述了workitem状态, 其中定义了initialed,running,suspened,active,terminated。
3. workflowContext 描述了一个workitem 的执行环境，包括engine、pkg、workflowpocess、参数表、返回值以及执行者、流程实例等。

对于单个活动执行控制:
1. joinMonitor: 对于预活动的处理
2. activityMonitor: 对活动的处理管理
3. LimitMonitor: 对受限活动的管理
4. Queued Activity 对需要人为的触发开始或者结束的活动的管理
5. RequiredActionType: 对于人工活动状态的定义和描述
6. repositorManager: 其中包含了procedure 库,application 库,participant

C  org.obe.engine.ejb
将workflow 发布成EJB 组件

D  org.obe.engine. persistence
包含数据保存的接口,以及实现将工作流数据写入数据库

E  org.obe.engine.util 对时间的管理

F  org.obe.engine.workitem
定义了application, noimplement,procedure 和subflow 四种workitem, 以及执行的方法

G  org.obe.event
描述了对于Activity,Error,PackageCollection 和WorkflowProcessInstance 的事件.
1. ActivityEvent 记录了该activity event,
2. ActivityListener(interface)用于侦听activityEvent,包含两个方法:activityStarted 和activityCompleted
3. ActivityListenerSupport:对于多个ActivityListener 的支持,较多个都记录在一起ErrorEvent 用于描述错误事件,当有任何错误发生的时候, 就会异步抛出该事件.ErrorListener(interface): 错误事件侦听, 当有错误发生的时候被调用.
4. ErrorListenerSupport:对于多个ErrorListener 的支持,将多个都记录在本类中,并提供了添加和删除操作
5. PackageCollectionEvent
6. WorkflowProcessInstanceEvent 如果激发了该事件,则将WorkflowProcessInstance加入


H  org.obe.runtime.application
对于流程执行期间的应用的调用处理,其中的组织关系:
1. ApplicationConnector(interface)=>ClassApplicationConnector 和debugApplicationConnector 以及RuntimeApplicationConnector 和XMLRPCApplicationConnector 四种，主要的作用是形成一个连接器，进行调用。
 ClassApplicationConnector 中根据classname 通过classloader 来载入类，通过invoke 方法来执行，执行类的main（调用外部应用）
 RuntimeApplicationConnector通过command 来调用应用。
 XMLRPCApplicationConnector 通过xmlrpc 的调用
2. ApplicationRepository=>BasicApplicationRepository 应用库用来获取运行时候的application connectors,在工作流执行的时候每一个活动应用的id 都被映射为一个特定的connector
3. Procedure(interface)=>PythonScriptProcedure 和ReflectionProcedure:是通过工作流引擎来直接执行的类,分别用来执行特定的Python 脚本和目标对象的特定方法
4. ProcedureRepository(interface)=> BasicProcedureRepository 用来获取执行的时候的procedure
5. Parameter 描述参数的值,参数名以及参数模式的对象
6. ReturnValue 描述返回值




I  org.obe.runtime.condition
对于流程执行过程中的条件的处理,其中
ConditionEvaluator(interface): 条件抽象
XpressionEvaluator(interface):
PythonConditionEvaluator: 利用Python 表达式求条件值


G  org.obe.runtime.participant
1. ParticipantConnector (interface):运行中参与者的工作列表;
2. ParticipantRepository(interface) :将每一个参与者和Participant- Connector 对应
3. BasicParticipantRepository 是一个实现类, 用map 表将两者做了映射org.obe.runtime.participant.connector
4. 该package 为具体的ParticipantConnector 实现类, 包括Human,Group 和workflowSystem.其中,RoundRobinGroup 是控制group 中的每一个人的分配.,是group 的扩展类


K  org.obe.wfxml
1. ExceptionCode 包含了所有的异常定义
2. ExceptionType 指明是fatal 异常,还是temporary 异常
3. Exception 是异常的描述,包括ExceptionCode 和ExceptionType

WFMessage 有messagePart,transport 和version ;
messagePart 包括WFMessageHeader 和messageBody
WFRequest 是一种请求的WFMessage, 在WFRequestHeader 中有相关的responseMessage的设定;
WFResponse 是一种回应的WFMessage ;
Transport 是消息的传输,将Exceptin 传送给特定的dialog

L  org.obe.wfxml.parser
包括从输入流中解析出输入请求和输出请求

M  org.obe.wfxml.transport
定义了dialog


3. obe test
obe test 是一个测试部分,包含两个类:TestObe 和Example,利用obeengine
和obecore 来进行流程的处理.

4. 流程处理过程
大致的处理过程如下：由于有两种执行方式asynch 和synch.这两种方式的区别在于synch 采用了ThreadReference 对于运行线程进行间管，而asynch 是直接运行线程.
首先定义一个workflowEngine，再定义一个package，利用xpdlParser 来解析xpdl 文件，由于package 是针对于业务流程xpdl 的描述而构造的类，所以可以将流程信息的各个部分利用obeCore 中定义的各个部分来进行初始化。然后将该package 加入到workflowEngine 中。workflowEngine 中用一个packages(list)来存储多个Package 类型。

Example usage:
1. WorkflowEngine engine = new WorkflowEngine();
2. XPDLParser parser = new Dom4JXPDLParser();
3. Package pkg = parser.parse(new FileInputStream("processdef.xml"));
4. engine.addPackage(pkg);
5. engine.executeSynch(pkg.getId(), "workflowID", new ArrayList());

WorkflowEngine 中对于流程的执行分别用了两种方法,

• engine.executeAsynch(packageId, processId, getParameters());
• engine.executeSynch(packageId, processId, getParameters());

\在executeSynch 中,是通过建立一个WorkflowProcessInstance 来执行流程,即
WorkflowEngine.createProcessInstance(packageId, workflowId,parameters);


1.在该方法createProcessInstance 中,首先根据packageId 在packages 中找到相应的package, 然后根据该package 和workflowID 来找到一个WorkflowProcess，
(public WorkflowProcessfindWorkflowProcessById(Package pkg, String workflowId))
如果package中workflow 返回为null,出错.在描述workprocess 中,使用一系列的list 来描述流程中的活动,条件,转移参数,参与者等等.

2.然后对于该package 和workflowprocess 进行检测
void validate(Package pkg, WorkflowProcess wp);主要是针对于时间的检测,从workflowprocess 的ProcessHeader 中提取出起始时间时间和结束时间以及持续时间.(如果将来做可扩展的话,可以在这里加入死锁的初步检测)

3. 然后构建一个workflow context:
WorkflowContext wc =new WorkflowContext(this, pkg, wp, parameters,new ArrayList(), getResponsibles(pkg, wp));
在context 中包含了engine,pkg,workflowprocess,以及workflowProcessInstance 参数表,返回值表,Responsibles 是在xml 中定义的<xpdl:RedefinableHeader>中包含的属性

4. 在WorkflowEngine 中对于一个WorkflowEngine 可以包含多个workflowProcessInstance, 也就是说, 可以创建多个线程的流程处理实例. 在workflowEngine 中用Hasp 来储存这些实例
下面就可以利用线程来启动执行流程.
首先通过WorkflowInstance 中的startProcess 方法来返回一个ThreadReference:
在startProcess()方法中, 通过创建一个WorkflowProcessRunner 线程来执行流程控制.

4.1、 对于WorkflowProcessRunner, 在初始化的时候, 有两个参数:
WorkflowProcessInstance 和WorkflowProcessInstanceListenerSupport.
在WorkflowProcessRunner 中实现了流程的执行控制（下面将专门进行分析）
WorkflowProcessInstanceListenerSupport 对于控制事件提供支持

4.2、WorkflowProcessRunner 的线程被创建和Start()后,开始执行run();同时还有一系列的方法,控制了流程的活动执行\活动转移等
// create the process thread
thread=new Thread(new WorkflowProcessRunner(this,
instanceListenerSupport),"PID-"+pid)// start the process thread
thread.start();

4.3、用ThreadReference 来对于执行线程进行管理
ThreadReference threadRef = processInstance.startProcess();
threadRef.join();
在startProcess()中创建和启动线程后,将threadReference = new ThreadReference(thread); return threadReference;
利用ThreadReference间接控制线程.

5. 流程中活动顺序的控制:
在流程执行的过程中,除了单纯的活动执行外,还有活动的顺序控制包括Split和and等. 在obe中,对于动态的过程实现是通过WorkFlowRunner(线程)实现的。大体的过程是这样的：
在workFlowProcess中已经建立了三个list,分别存放所有的activities, participants,transitions.

5.1、首先建立三个HashMap,将activity存入HashMap;
然后初始化开始活动:
如果已经明确设定了初始活动,则就直接设定;否则,将list中的id=0的活动设为开始活动:

5.2、然后执行void executeActivity(Activity activity) throws Exception
在该方法中:首先从对于activity对象中抽出对于activity的implication, 得到activity的开始模式和结束模式,根据是MANUAl还是AUTOMATIC采取不同的处理(对于人工开始和结束的方式单独分析)
然后执行.执行完毕后,根据技术模式来结束当前活动,执行
executeOutgoingTransitions(activity, TransitionLoopType.NOLOOP)

5.3、在executeOutgoingTransitions中:
首先找到所有的当前活动的后继的转移,然后找到split的类型(XOR还AND),并执行该(split);
executeSplit(transitionsUnion, splitType);如果当前的活动节点是终极节点,
则返回;如果改活动的后继转移是唯一的,则直接转移执行下一个活动.

5.4、executeSplit(transitionsUnion, splitType);对于当前活动的后继所有转移逐个进行分析,

if(splitType == SplitType.AND){
log.info("AND split; execute in new thread");
// execute the activity in a parallel thread
Thread t = new Thread(new ActivityRunner(
transition.getTo(), this));
// find the parent thread reference
ThreadReference threadReference = new ThreadReference(t,threadReferenceParent);
threadReferenceParent.getChildren().add(threadReference);
// start the thread
t.start();
} else if(splitType == SplitType.XOR){
log.info("XOR split; execute in same thread");
executeActivity(transition.getTo());
return;
}

评论

5 楼 timeson 2009-12-30  

你从cvs下载的是新版本的obe，基于应用服务器的（jboss）
我这个版本的obe，是最老的版本，纯java代码的
新版obe是在老版本obe基础上作了重构，而且2个版本的作者也不一样

老版本的obe，直接在开发环境中就可以运行，不需要部署在某个容器内

4 楼 wu2leng 2009-12-25  

最近，一直在看OBE。从CVS上下的源码，可是不怎么会部署。老是发布到JBOSS的时候出现错误。我想请问，博主有详细部署obe的说明吗？我用的windows+mysql5.1+jboss4.2.3
还有，博主您上传的这个old—OBE，我老是编译TEST.xml的时候出错。晕啊 。。。。。
现在不知道怎么往下做了。。呜呜。。。

3 楼 wu2leng 2009-12-12  

非常希望博主能吧详细的编译配置图解写成博客，赞赞赞！！！

2 楼 timeson 2009-12-08  

在包中的classes目录下就有这个xml文件
；）

1 楼 wu2leng 2009-12-03  

请问有OBE中的那个Test.XML例子吗？？能给我发个吗？谢谢了