JBPM4异步处理框架浅析

在介绍JBPM异步框架我们先说点点OOP的概念,在我们要为一个类不增加方法的情况下为此类增加功能有两种方法，一个是继承,一个是组合。继承以及组合都是提高代码复用的主要方式。组合就是让主类拥有一个组合类的引用。这种技巧在面向对象程序编写中是非常常见的。
Jbpm的异步框架模型


                           图：jobExecutor框架图
由此我们可以看出jbpm的异步框架模型,分为两个重要的实现，
1、 执行活动发送异步消息即在jbpm_job中写一条数据。


                   图：异步活动发送异步消息入库
在executionImpl调用execute方法的时候,当执行到原子操作(ExecutionActivity)时，先判断当前活动,若是异步的则执行上图逻辑，即生成异步消息并入库表JBPM4_JOB的同时,在事务中注册一执行同步的通知。即事务结束即执行afterCompletion（）方法。通知处于等待状态的dispathcerThread线程启动。
2、 扫描jbpm_job中的数据，并生成对应的任务（Runnable）提交到线程池中，自动运行。
要了解JBPM对异步消息的处理框架,我们先要弄明白java.util.concurrent的Executor框架的核心思想。
Executor框架核心


                     图：ThreadPoolExector类的核心方法
ThreadPoolExector是如何管理和分配线程的呢，首先我们调用submit()方法将我们定义好的任务提交到线程池中，然后,ThreadPoolExector类将调用execute方法，当然该方法将回调Runnable的start方法。至于如何关闭线程，则需要调用ThreadPoolExector的shutdown（）方法。至于线程的创建以及分配则由ThreadPoolExector全权代理，我们勿需关心。
        对于executor框架的扩展,JBPM主要是包装了ThreadPoolExecutor类的各个方法。
     JBPM主要定义了JobExecutor和DispatcherThread两个类来完成异步消息处理。DispatherThread类主要实现功能如下：
1、 查询独占式job列表
hql如下：
  

select job
     from org.jbpm.pvm.internal.job.JobImpl as job
     where job.lockOwner is null
       and job.processInstance = :processInstance 
       //可执行的Job
       and job.isExclusive = true 
       and job.retries > 0 
//任务状态为非暂停
       and job.state != 'suspended'
       and ( (job.duedate is null)
             or (job.duedate <= :now) 
           )

将job列表中的job生成JobParcel并提交到线程池中。
相关代码：

jobExecutor.getThreadPool().submit(
        new JobParcel(jobExecutor.getCommandExecutor(), jobDbids));

JobExecutor类主要实现功能如下：
1、 初始化线程池

threadPool = new ThreadPoolExecutor(nbrOfThreads, 
                                          nbrOfThreads, 
                                          0L, 
                                          TimeUnit.MILLISECONDS,
                                          new ArrayBlockingQueue<Runnable>(nbrOfThreads), 
                                          JobRejectionHandler.INSTANCE);

几个参数的具体含义如下：
1、 corePoolSize,maxinumPoolSize设置。
  当线程数小于corePoolSize时候，则创建新线程来处理请求，即使线程池中的线程是空闲的。当线程数大于corePoolSize而小于maxinumPoolsize，则仅当队列满时创建新的线程。当corePoolsize与maxinumPoolSize相同，则创建了固定线程池。当maxinumPoolSize为Integer.MAX_VALUE则创建的是一个无界线程池。允许线程池创建任意数量的线程来处理并发任务。
因此，如何调整线程池大小，以期得到最好的性能。
1） 调整corePoolSize
2） 调整队列大小
2、 KeepliveTime设置
  定义线程池中多余corePoolSize的空闲线程存活的时间。当空闲多余这个时间的时候，线程池就回收这些线程。
3、 TimeUnit.MILLISECONDS定义线程空闲时间的单位。
4、 BlockQueue可以用于保存和传输提交的任务。
可以使用此队列与线程池进行交互。
1）如果运行的线程少于 corePoolSize，则 Executor 始终首选添加新的线程，而不进行排队。
2）如果运行的线程等于或多于 corePoolSize，则 Executor 始终首选将请求加入队列，而不添加新的线程。
3）如果无法将请求加入队列，则创建新的线程，除非创建此线程超出 maximumPoolSize，在这种情况下，任务将被拒绝
排队有三种通用策略：
直接提交。工作队列的默认选项是 SynchronousQueue，它将任务直接提交给线程而不保持它们。在此，如果不存在可用于立即运行任务的线程，则试图把任务加入队列将失败，因此会构造一个新的线程。此策略可以避免在处理可能具有内部依赖性的请求集合时出现锁定。直接提交通常要求无界 maximumPoolSizes 以避免拒绝新提交的任务。当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。
无界队列。使用无界队列（例如，不具有预定义容量的 LinkedBlockingQueue）将导致在所有 corePoolSize 线程都忙的情况下将新任务加入队列。这样，创建的线程就不会超过 corePoolSize。（因此，maximumPoolSize 的值也就无效了。）当每个任务完全独立于其他任务，即任务执行互不影响时，适合于使用无界队列；例如，在 Web 页服务器中。这种排队可用于处理瞬态突发请求，当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。
有界队列。当使用有限的 maximumPoolSizes 时，有界队列（如 ArrayBlockingQueue）有助于防止资源耗尽，但是可能较难调整和控制。队列大小和最大池大小可能需要相互折衷：使用大型队列和小型池可以最大限度地降低 CPU 使用率、操作系统资源和上下文切换开销，但是可能导致人工降低吞吐量。如果任务频繁阻塞（例如，如果它们是 I/O 边界），则系统可能为超过您许可的更多线程安排时间。使用小型队列通常要求较大的池大小，CPU 使用率较高，但是可能遇到不可接受的调度开销，这样也会降低吞吐量。
5、 被拒绝任务处理
拒绝任务的处理,threadPoolExecutor提供以下几种策略：
1） AbortPolicy 策略
处理程序遭到拒绝，将抛出异常RejectExecutionException
2） CallerRunsPolicy
线程调用运行该任务的execute本身。此策略提供简单的反馈控制机制，能够减缓新任务的提交速度。
3） DiscardPolicy 不能执行任务将被删除
4） DiscardOldestPolicy 如果执行程序尚未关闭，则位于工作队列的头部的任务将被删除。然后重新执行程序。