quartzschedulerthread.java

来自「定时器开源项目, 相对于 jcrontab, Quartz 算是更完整的一个项目」· Java 代码 · 共 450 行 · 第 1/2 页

            Trigger trigger = null;
            boolean idleWait = true;

            try {
                trigger = qsRsrcs.getJobStore().acquireNextTrigger(
                        ctxt);
                lastAcquireFailed = false;
            } catch (JobPersistenceException jpe) {
                if(!lastAcquireFailed)
                    qs.notifySchedulerListenersError(
                        "An error occured while scanning for the next trigger to fire.",
                        jpe);
                lastAcquireFailed = true;
            }
            catch (RuntimeException e) {
                if(!lastAcquireFailed)
                    getLog().error("quartzSchedulerThreadLoop: RuntimeException "
                            +e.getMessage(), e);
                lastAcquireFailed = true;
            }

            if (trigger != null) {
                long now = System.currentTimeMillis();

                long triggerTime = trigger.getNextFireTime().getTime();
                long timeUntilTrigger = triggerTime - now;
                long spinInterval = 10;

                if (timeUntilTrigger <= idleWaitTime) {
                    // this looping may seem a bit silly, but it's the
                    // current work-around
                    // for a dead-lock that can occur if the Thread.sleep()
                    // is replaced with
                    // a obj.wait() that gets notified when the signal is
                    // set...
                    // so to be able to detect the signal change without
                    // sleeping the entire
                    // timeUntilTrigger, we spin here... don't worry
                    // though, this spinning
                    // doesn't even register 0.2% cpu usage on a pentium 4.
                    int numPauses = (int) (timeUntilTrigger / spinInterval);
                    while (numPauses >= 0 && !signaled) {

                        try {
                            Thread.sleep(spinInterval);
                        } catch (InterruptedException ignore) {
                        }

                        now = System.currentTimeMillis();
                        timeUntilTrigger = triggerTime - now;
                        numPauses = (int) (timeUntilTrigger / spinInterval);
                    }
                    if (signaled) {
                        try {
                            qsRsrcs.getJobStore().releaseAcquiredTrigger(
                                    ctxt, trigger);
                        } catch (JobPersistenceException jpe) {
                            qs.notifySchedulerListenersError(
                                    "An error occured while releasing trigger '"
                                            + trigger.getFullName() + "'",
                                    jpe);
                            // db connection must have failed... keep
                            // retrying until it's up...
                            releaseTriggerRetryLoop(trigger);
                        } catch (RuntimeException e) {
                            getLog().error(
                                "releaseTriggerRetryLoop: RuntimeException "
                                +e.getMessage(), e);
                            // db connection must have failed... keep
                            // retrying until it's up...
                            releaseTriggerRetryLoop(trigger);
                        }
                        signaled = false;
                        continue;
                    }

                    // set trigger to 'executing'
                    TriggerFiredBundle bndle = null;

                    try {
                        bndle = qsRsrcs.getJobStore().triggerFired(ctxt,
                                trigger);
                    } catch (SchedulerException se) {
                        qs.notifySchedulerListenersError(
                                "An error occured while firing trigger '"
                                        + trigger.getFullName() + "'", se);
                    }

                    if (bndle == null) {
                        try {
                            qsRsrcs.getJobStore().releaseAcquiredTrigger(ctxt,
                                    trigger);
                        } catch (SchedulerException se) {
                            qs.notifySchedulerListenersError(
                                    "An error occured while releasing trigger '"
                                            + trigger.getFullName() + "'", se);
                            // db connection must have failed... keep retrying
                            // until it's up...
                            releaseTriggerRetryLoop(trigger);
                        }
                        continue;
                    }

                    JobRunShell shell = null;
                    try {
                        shell = qsRsrcs.getJobRunShellFactory()
                                .borrowJobRunShell();

                        shell.initialize(qs, bndle);
                    } catch (SchedulerException se) {
                        try {
                            qsRsrcs.getJobStore().releaseAcquiredTrigger(ctxt,
                                    trigger);
                        } catch (SchedulerException se2) {
                            qs.notifySchedulerListenersError(
                                    "An error occured while releasing trigger '"
                                            + trigger.getFullName() + "'", se2);
                            // db connection must have failed... keep retrying
                            // until it's up...
                            releaseTriggerRetryLoop(trigger);
                        }
                        continue;
                    }

                    qsRsrcs.getThreadPool().runInThread(shell);

                    idleWait = false;
                } else {
                    //put the trigger back into the queue so it may be
                    // executed again in future
                    try {
                        qsRsrcs.getJobStore().releaseAcquiredTrigger(ctxt,
                                trigger);
                    } catch (JobPersistenceException jpe) {
                        qs.notifySchedulerListenersError(
                                "An error occured while releasing trigger '"
                                        + trigger.getFullName() + "'", jpe);
                        // db connection must have failed... keep retrying
                        // until it's up...
                        releaseTriggerRetryLoop(trigger);
                    } catch (RuntimeException e) {
                        getLog().error(
                            "releaseTriggerRetryLoop: RuntimeException "
                            +e.getMessage(), e);
                        // db connection must have failed... keep retrying
                        // until it's up...
                        releaseTriggerRetryLoop(trigger);
                    }
                    idleWait = true;
                }
            }

            // this looping may seem a bit silly, but it's the current
            // work-around
            // for a dead-lock that can occur if the Thread.sleep() is replaced
            // with
            // a obj.wait() that gets notified when the signal is set...
            // so to be able to detect the signal change without sleeping the
            // entier
            // getRandomizedIdleWaitTime(), we spin here... don't worry though,
            // the
            // CPU usage of this spinning can't even be measured on a pentium
            // 4.
            long now = System.currentTimeMillis();
            long waitTime = now + getRandomizedIdleWaitTime();
            long timeUntilContinue = waitTime - now;
            long spinInterval = 10;
            int numPauses = (int) (timeUntilContinue / spinInterval);

            while (idleWait && numPauses > 0 && !signaled) {

                try {
                    Thread.sleep(10L);
                } catch (InterruptedException ignore) {
                }

                now = System.currentTimeMillis();
                timeUntilContinue = waitTime - now;
                numPauses = (int) (timeUntilContinue / spinInterval);
            }
            signaled = false;
        } // loop...

        // drop references to scheduler stuff to aid garbage collection...
        qs = null;
        qsRsrcs = null;
    }

    public void releaseTriggerRetryLoop(Trigger trigger) {
        int retryCount = 0;
        try {
            while (!halted) {
                try {
                    Thread.sleep(getDbFailureRetryInterval()); // retry every N
                    // seconds (the db
                    // connection must
                    // be failed)
                    retryCount++;
                    qsRsrcs.getJobStore().releaseAcquiredTrigger(ctxt, trigger);
                    retryCount = 0;
                    break;
                } catch (JobPersistenceException jpe) {
                    if(retryCount % 4 == 0)
                        qs.notifySchedulerListenersError(
                            "An error occured while releasing trigger '"
                                    + trigger.getFullName() + "'", jpe);
                } catch (RuntimeException e) {
                    getLog().error("releaseTriggerRetryLoop: RuntimeException "+e.getMessage(), e);
                } catch (InterruptedException e) {
                    getLog().error("releaseTriggerRetryLoop: InterruptedException "+e.getMessage(), e);
                }
            }
        } finally {
            if(retryCount == 0)
                getLog().info("releaseTriggerRetryLoop: connection restored.");
        }
    }

    public static Log getLog() {
        return LogFactory.getLog(QuartzSchedulerThread.class);
    }

} // end of QuartzSchedulerThread