写一个简单的工作流（四）资源的处理

昨天晚上搞到深夜，终于将资源模块搞定。到今天已经完成的功能包括： 1.四种基本路由：顺序、选择、并行、循环 2.流程定义文件和系统配置文件的读取和解析 3.使用内存作为流程数据和案例数据存储的MemoryWorkFlowDAO的开发 4.资源模块的开发 5.并发情况下的正确性测试等     计划中的功能： 1.一个GUI的流程定义工具，这个不急，也还没想好用什么做，web还是桌面？ 2.各个数据库版本的WorkFlowDAO的开发，将流程数据和案例数据保存在数据库中。 3.更多的测试和example试验。     回到资源这个概念，工作流中工作项（work item）的由资源来驱动的，这个资源(resource)可能是用户、角色、定时时间或者某个事件消息。在标准petri网中，工作项对应于transition(变迁），变迁都是自动的，不需要所谓资源来驱动，显然，这与工作流系统不同。具体到insect workflow(我取的名字，小巧之意），每个transition都有一个resource，用于驱动自身的firing，所有的resource都实现Resource接口： public   interface  Resource  extends  Serializable {      public   void  start(Transition transition, Token token, Object args);                  public  ResourceType getType();      public   long  getId(); }     每个资源都有一个类型，以及这个类型中独一无二的id，start方法用于驱动transtion的firing。一般情况下，你不需要实现这个接口，只要继承这个接口的抽象实现类AbstractResource，AbstractResource的start方法默认实现是首先调用模板方法doAction（稍后解释）,然后检查触发条件，如果通过就直接调用transition的fire方法： public   abstract   class  AbstractResource  implements  Resource {                public   void  start(Transition transition, Token token, Object args) {         doAction(transition, token, args);          if  (transition.getCondition()  !=   null                  &&   ! transition.getCondition().check(token))              throw   new  ConditionException(transition.getName()                      +   "  transition没有满足触发条件 " );         transition.fire(token, args);     }      public   abstract   void  doAction(Transition transition, Token token,             Object args) ;         }     Transtion类的fire方法有三个操作组成：从输入库所移走token，往输出库所放入token，回调handler：     public   void  fire(Token token, Object args) {         removeTokenFromInputs(token);         addTokenToOutputs(token);         invokeHandler(token, args);     }     那么具体的资源显然要实现AbstractResource中的doAction抽象方法，系统内置了五种资源：自动资源（AutoResource）、用户(User)、用户组(Group)、定时器(TimerResource)和事件监听器（ObserverResource）。显然，AutoResource、User和Group的doAction方法不需要做任何事情： public   class  User  extends  AbstractResource {      protected  Group group;                  @Override      public   void  doAction(Transition transition, Token token, Object arg) {     } }         而TimerResource就需要做特殊处理了，比如我们要达到这样的效果：节点1状态已经处于就绪，可以被触发，可我们希望在就绪后延迟半分钟再触发，或者在晚上10点触发等等。这样的定时需求很常见，我采用了jdk5引入的ScheduledExecutorService来处理。系统中启动这样一个线程池，每个类似上面的请求都提交给这个线程池来处理，那么TimerResource就需要进行相应的修改： public   abstract   class  TimerResource  extends  AbstractResource {      protected   int  pool_size;      protected   static  ScheduledExecutorService scheduledExecutorService;     @Override      public   long  getId() {          //  TODO Auto-generated method stub          return  Common.TIMER_RESOURCE_ID;     }      public  TimerResource() {          this .pool_size  =   5 ;         scheduledExecutorService  =  Executors.newScheduledThreadPool(pool_size);     }      public   static   void  shutdownPool() {          if  (scheduledExecutorService  !=   null )             scheduledExecutorService.shutdown();     }      public   final   void  start(Transition transition, Token token, Object args)              throws  InterruptedException {          if  (transition.getCondition()  !=   null                  &&   ! transition.getCondition().check(token))              throw   new  ConditionException(transition.getName()                      +   "  transition没有满足触发条件 " );         transition.removeTokenFromInputs(token);         doAction(transition, token, args);     }      protected   class  ChangeRunner  implements  Runnable {          private  Transition transition;          private  Token token;          private  Object[] args;          public  ChangeRunner(Transition transition, Token token, Object args) {              this .transition  =  transition;              this .token  =  token;              this .args  =  args;         }          public   void  run() {              if  (transition.getCondition()  !=   null                      &&   ! transition.getCondition().check(token))                  throw   new  ConditionException(transition.getName()                          +   "  transition没有满足触发条件 " );             transition.addTokenToOutputs(token);             Object real_args[]  =   new  Object[args.length  -   2 ];              for  ( int  i  =   0 ; i  <  real_args.length; i ++ )                 real_args[i]  =  args[i  +   2 ];             transition.invokeHandler(token, real_args);              try  {                  //  回调                 ((WorkFlowAlgorithm) args[ 1 ]).enabledTraversing(token                         .getWorkFlow());                 ((WorkFlowManager) args[ 0 ]).doAction(token.getId());             }  catch  (InterruptedException e) {                 Thread.currentThread().interrupt();             }         }     } }     注意到，start方法不再是直接调用transition的fire方法，而仅仅是进行了第一步操作：移除输入库所的place防止重复提交。后两步操作都延迟到了提交给线程池的任务中，也就是代码中的 ChangeRunner类中的run方法。例如TimerResource的子类DelayTimerResource用于处理延迟的触发，doAction就像这样： public   class  DelayTimerResource  extends  TimerResource {          @Override      public   void  doAction(Transition transition, Token token, Object args) {         scheduledExecutorService.schedule( new  ChangeRunner(transition, token,                 args),  this .delay,  this .timeUnit);     } }     延迟的时间，时间单位这些信息都可以在流程定义文件中设置。事件监听器资源与此类似，ObserverResource实现了java.util.Observer接口，往输出库所放入token和回调handler两步操作被放在了update方法中提供给Subject回调。    文章转自庄周梦蝶  ，原文发布时间2007-10-13