OFBiz中JOB的运行机制

OFBiz执行后台任务的类在org.ofbiz.service.job中。
JobPoller和JobInvoker是主要的两个类，一个负责查询可以执行的Job，另一个执行Job任务。Job类图如下所示。

1.Job轮询
创建JobManager时，会创建JobPoller的一个实例。JobPoller实现了Runnable接口，以此创建线程后 通过JobManager一直轮询是否有Job需要执行，如果有奖将其放入队列中。

 1     public synchronized void run() {  2         try {  3             // wait 30 seconds before the first poll  4             java.lang.Thread.sleep(30000);  5         } catch (InterruptedException e) {  6         }  7         while (isRunning) {  8             try {  9                 // grab a list of jobs to run. 10                 List<Job> pollList = jm.poll(); 11                 //Debug.logInfo("Received poll list from JobManager [" + pollList.size() + "]", module); 12  13                 for (Job job : pollList) { 14                     if (job.isValid()) { 15                         queueNow(job); 16                         //Debug.logInfo("Job [" + job.getJobId() + "] is queued", module); 17                     } 18                 } 19                 // NOTE: using sleep instead of wait for stricter locking 20                 java.lang.Thread.sleep(pollWaitTime()); 21             } catch (InterruptedException e) { 22                 Debug.logError(e, module); 23                 stop(); 24             } 25         } 26     } 27 

queueNow方法将要执行job放入到队列中，如果队列中的等待执行的job数量很多，那么就创建一定数量的线程执行这些job。

 1    public void queueNow(Job job) {  2         synchronized (run) {  6             run.add(job);  7         }  8         if (Debug.verboseOn()) Debug.logVerbose("New run queue size: " + run.size(), module);  9         if (run.size() > pool.size() && pool.size() < maxThreads()) { 10             synchronized (pool) { 11                 if (run.size() > pool.size() && pool.size() < maxThreads()) { 12                     int calcSize = (run.size() / jobsPerThread()) - (pool.size()); 13                     int addSize = calcSize > maxThreads() ? maxThreads() : calcSize; 14  15                     for (int i = 0; i < addSize; i++) { 16                         JobInvoker iv = new JobInvoker(this, invokerWaitTime()); 17                         pool.add(iv); 18                     } 19                 } 20             } 21         } 22     }

JobInvoker就是执行的线程，它从queue中取job并执行。JobInvoker线程不是一直运行下去，运行的时间长度超过一定的值（见 serviceengine.xml中ttl的值）线程就会停止并从pool中删除。JobInvoker的run方法中job.exec()执行具体的 任务。
2.Job执行
Job类都有一个exec方法，用户执行Job的service。如GenericServiceJob中的exec方法如下：

 1     public void exec() throws InvalidJobException {  2         init();  3   4         // no transaction is necessary since runSync handles this  5         try {  6             // get the dispatcher and invoke the service via runSync -- will run all ECAs  7             LocalDispatcher dispatcher = dctx.getDispatcher();  8             Map<String, Object> result = dispatcher.runSync(getServiceName(), getContext());  9  10             // check for a failure 11             boolean isError = ModelService.RESPOND_ERROR.equals(result.get(ModelService.RESPONSE_MESSAGE)); 12             if (isError) { 13                  String errorMessage = (String) result.get(ModelService.ERROR_MESSAGE); 14                  this.failed(new Exception(errorMessage)); 15             } 16  17             if (requester != null) { 18                 requester.receiveResult(result); 19             } 20  21         } catch (Throwable t) { 22             // pass the exception back to the requester. 23             if (requester != null) { 24                 requester.receiveThrowable(t); 25             } 26  27             // call the failed method 28             this.failed(t); 29         } 30  31         // call the finish method 32         this.finish(); 33     } 34 

在执行service执行，有一个init方法，在PersistedServiceJob类中init方法主要是生成下一个执行的任务，如果有的话。也 即是说每一个job是由当时执行的这个job生成的，根据是什么呢？主要是两个变量：tempExprId和 maxRecurrenceCount，init方法中：     

 1     TemporalExpression expr = null;  2     ……  3   4     if (expr == null && UtilValidate.isNotEmpty(job.getString("tempExprId"))) {  5             try {  6                 expr = TemporalExpressionWorker.getTemporalExpression(this.delegator, job.getString("tempExprId"));  7             } catch (GenericEntityException e) {  8                 throw new RuntimeException(e.getMessage());  9             } 10         } 11 

TemporalExpressionWorker里面有一个makeTemporalExpression方法很重要，从这个方法可以知道怎么配置 TemporalExpression实体数据了，当然要结合TemporalExpressions类，里面定义了各种配置的细节。
tempExprTypeId有如下几种：
DateRange DayInMonth DayOfMonthRange DayOfWeekRange Difference Frequency Intersection MonthRange TimeOfDayRange Union
比如如果希望服务只执行一次，可以如下配置：     <TemporalExpression tempExprId="RUNONCE" tempExprTypeId="FREQUENCY" integer1="1" integer2="1"/>     <JobSandbox jobId="CurrencyRateSynAll" jobName="Currency Rate SynAll" runTime="2010-02-26 09:38:00.000" serviceName="currencyRateSynAll" poolId="pool" runAsUser="system" tempExprId="RUNONCE" maxRecurrenceCount="0"/>
maxRecurrenceCount="0" 表示，不重复。tempExprTypeId="FREQUENCY" integer1="1" integer2="1"表示一年执行一次。所以总共执行一次就结束了。
每天都执行可以这样配置：
<TemporalExpression tempExprId="MIDNIGHT_DAILY" tempExprTypeId="TIME_OF_DAY_RANGE" string1="20:00:00" string2="20:00:00"/>     <JobSandbox jobId="MailNotification" jobName="Mail Notification Job" runTime="2010-02-25 18:00:00.000" serviceName="mailNotificantion" poolId="pool" runAsUser="system" tempExprId="MIDNIGHT_DAILY" maxRecurrenceCount="-1"/>
maxRecurrenceCount="-1"表示无限循环下去。tempExprId="MIDNIGHT_DAILY" tempExprTypeId="TIME_OF_DAY_RANGE" string1="20:00:00" string2="20:00:00"/>表示每天晚上八点执行。
每个月一次任务可以如下配置：
<TemporalExpression tempExprId="ONCEINMONTH" tempExprTypeId="FREQUENCY" date1="2010-02-26 11:05:00.000" integer1="2" integer2="1"/>     <JobSandbox jobId="CurrencyRateSyn" jobName="Currency Rate Syn" runTime="2010-02-26 11:05:00.000" serviceName="currencyRateSyn" poolId="pool" runAsUser="system" tempExprId="ONCEINMONTH" maxRecurrenceCount="-1"/>
tempExprTypeId="FREQUENCY" date1="2010-02-26 11:05:00.000" integer1="2" integer2="1"表示每月一次，时间就是date1定义的时间，如果没用定义date1，那么就是当前时间。
这里的配置相当灵活，好好掌握。

 

仅供个人收藏 转载自：http://www.blogjava.net/persister/archive/2010/02/26/313974.html