java - Quartz Scheduler抢先触发优先级

Question

是否可以实现某种与触发优先级相关的“先发制人”行为？

我的意思是，我想要一个高优先级触发器来中断当前正在运行的低优先级作业，并在其位置上运行。

我想更进一步，不仅比较同一作业的触发优先级，还比较尝试处理同一“资源”的不同作业的触发优先级，不是同时比较，而是在重叠的时间(假设“工作”需要时间才能完成)。

我没有找到任何“开箱即用”的东西。有人曾经实现过类似的东西吗？

Answer 1

这是我迄今为止的解决方案(已删除导入)。有任何警告或改进吗？

//
// AN EXAMPLE JOB CLASS
//

public class DevJob implements InterruptableJob {
    private final transient Logger log = LoggerFactory.getLogger(getClass());

    AtomicReference<Thread> jobThreadHolder = new AtomicReference<Thread>();

    @Override
    public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
        String triggerName = jobExecutionContext.getTrigger().getKey().toString();
        String jobName = jobExecutionContext.getJobDetail().getKey().toString();
        log.debug("Executing Job {}-{} ", triggerName, jobName);
        jobThreadHolder.set(Thread.currentThread());
        JobDataMap jobDataMap = jobExecutionContext.getJobDetail().getJobDataMap();
        int duration = jobDataMap.getInt("duration");
        try {
            log.debug("Job {}-{} works for {}s...", triggerName, jobName, duration);
            Thread.sleep(duration*1000);
        } catch (InterruptedException e) {
            log.debug("Job {}-{} interrupted", triggerName, jobName);
            PreemptiveVolatileQueueQuartzListener.setInterrupted(jobExecutionContext);
        } finally {
            log.debug("Job {}-{} terminates", triggerName, jobName);
        }
    }

    @Override
    public void interrupt() throws UnableToInterruptJobException {
         Thread thread = jobThreadHolder.getAndSet(null);
         if (thread != null) {
             thread.interrupt();
         }
    }
}


//
// IMPLEMENTATION FOR JOB PREEMPTION
//

/**
 * This class implements a priority policy for jobs in the same thread group.
 * When a new job starts, it will check if another job in the same thread group is already running.
 * In such a case it compares trigger priorities. The job with lower priority is put in a wait queue for the thread group,
 * but only if another instance with the same jobKey is not in the queue already.
 * When the running job terminates, a new job is pulled from the queue based on priority and timestamp: for equal
 * priorities, the older one is executed.
 * If a job has been interrupted MAX_RESCHEDULINGS times, it will ignore any further interruptions.
 * A job must implement InterruptableJob and periodically call checkInterruptRequested() if it can be interrupted by a
 * higher priority job; it could ignore interruptions, in which case the higher priority job will execute only after 
 * its natural termination. 
 */
public class PreemptiveVolatileQueueQuartzListener implements JobListener, TriggerListener {
    private final transient Logger log = LoggerFactory.getLogger(getClass());

    // The number of times that a low priority job can be preempted by any high priority job before it ignores preemption
    private static final int MAX_RESCHEDULINGS = 20; 

    // This map holds the pointer to the current running job and its deferred queue, for a given thread group
    private Map<String, RunningJobHolder> runningJobs = new HashMap<>(); // triggerGroup -> RunningJob

    private static final String INTERRUPTED_FLAG = "PREEMPT_INTERRUPTED";
    private static final String INTERRUPTREQUESTED_FLAG = "PREEMPT_INTERRUPTREQUESTED";
    static final String JOB_ORIG_KEY = "PREEMPT_JOBORIGKEY";

    /**
     * Call this method to notify a job that an interruption has been requested. It should tipically be called
     * in the InterruptableJob.interrupt() method. The job will then have to programmatically check this flag with checkInterruptRequested()
     * and exit if the result is true.
     */
    public final static void requestInterrupt(JobExecutionContext jobExecutionContext) {
        jobExecutionContext.getJobDetail().getJobDataMap().put(INTERRUPTREQUESTED_FLAG, true);
    }

    /**
     * Call this method in a job to check if an interruption has been requested. If the result is true, the "interrupted" flag
     * will be set to true and the job should exit immediately
     * because it will be rescheduled after the interrupting job has finished.
     * @param jobExecutionContext can be null if the check should not be performed
     * @return true if the interruption was requested
     */
    public final static boolean checkInterruptRequested(JobExecutionContext jobExecutionContext) {
        boolean result = false;
        if (jobExecutionContext!=null) {
            try {
                result = jobExecutionContext.getJobDetail().getJobDataMap().getBoolean(INTERRUPTREQUESTED_FLAG);
            } catch (Exception e) {
                // Ignore, stay false
            }
            if (result) {
                setInterrupted(jobExecutionContext);
            }
        }
        return result;
    }

    /**
     * Call this method in a job when catching an InterruptedException if not rethrowing a JobExecutionException
     * @param jobExecutionContext
     */
    public final static void setInterrupted(JobExecutionContext jobExecutionContext) {
        jobExecutionContext.getJobDetail().getJobDataMap().put(INTERRUPTED_FLAG, true);
    }

    private final boolean isInterrupted(JobExecutionContext jobExecutionContext) {
        try {
            return true==jobExecutionContext.getJobDetail().getJobDataMap().getBoolean(INTERRUPTED_FLAG);
        } catch (Exception e) {
            return false;
        }
    }

    private final void clearInterrupted(JobExecutionContext jobExecutionContext) {
        jobExecutionContext.getJobDetail().getJobDataMap().remove(INTERRUPTREQUESTED_FLAG);
        jobExecutionContext.getJobDetail().getJobDataMap().remove(INTERRUPTED_FLAG);
    }

    /**
     * This method decides if a job has to start or be queued for later.
     */
    @Override
    public boolean vetoJobExecution(Trigger startingTrigger, JobExecutionContext startingJobContext) {
        log.debug("Calculating veto for job {}", makeJobString(startingTrigger));
        boolean veto = false;
        String preemptedGroup = startingTrigger.getKey().getGroup();
        synchronized (runningJobs) {
            veto = calcVeto(startingTrigger, preemptedGroup, startingJobContext);
        }
        log.debug("veto={} for job {}", veto, makeJobString(startingTrigger));
        return veto;
    }

    private boolean calcVeto(Trigger startingTrigger, String preemptedGroup, JobExecutionContext startingJobContext) {
        final boolean VETO = true;
        final boolean NOVETO = false;
        int startingJobPriority = startingTrigger.getPriority();
        RunningJobHolder runningJobHolder = runningJobs.get(preemptedGroup);
        if (runningJobHolder==null) {
            // No conflicting job is running - just start it
            runningJobHolder = new RunningJobHolder();
            runningJobs.put(preemptedGroup, runningJobHolder);
            PrioritizedJob newJob = runningJobHolder.setActiveJob(startingJobPriority, startingTrigger, startingJobContext);
            log.debug("Starting new job {} with nothing in the same group", newJob);
            return NOVETO;
        }

        // Check that the current job isn't a job that has just been pulled from the queue and activated
        boolean sameTrigger = startingTrigger.equals(runningJobHolder.activeJob.trigger);
        if (sameTrigger) {
            // runningJobHolder.activeJob has been set in triggerComplete but the trigger didn't fire until now
            log.debug("Starting trigger {} is the same as the active one", startingTrigger.getKey());
            return NOVETO; 
        }

        // Check that the starting job is not already running and is not already queued, because we don't want
        // jobs to accumulate in the queue (a design choice)
        if (runningJobHolder.isInHolder(startingTrigger)) {
            log.debug("Starting job {} is queued already (maybe with a different trigger)", makeJobString(startingTrigger));
            return VETO;
        }

        // A job for this triggerGroup is already running and is not the same as the one trying to start and is not in the queue already.
        // The starting job is therefore queued, ready to be started, regardless of the priority.
        PrioritizedJob newJob = runningJobHolder.queueJob(startingJobPriority, startingTrigger, startingJobContext);
        log.debug("New job {} queued", newJob);
        printQueue(runningJobHolder); // Debug

        if (startingJobPriority>runningJobHolder.activeJob.priority) {
            if (runningJobHolder.activeJob.reschedulings >= MAX_RESCHEDULINGS) {
                // When a job has been preempted too many times, it is left alone even if at lower priority
                log.debug("New job {} does not interrupt job {} of lower priority because its reschedulings are {}", newJob, runningJobHolder.activeJob, runningJobHolder.activeJob.reschedulings);
            } else {
                // The starting job has a higher priority than the current running job, which needs to be interrupted.
                // The new job will take the place of the running job later, because it has been added to the queue already.
                // If the running job doesn't react to the interruption, it will complete normally and let the next job in 
                // the queue to proceed.
                log.debug("New job {} interrupts job {} of lower priority", newJob, runningJobHolder.activeJob);
                try {
                    Scheduler scheduler = startingJobContext.getScheduler();
                    scheduler.interrupt(runningJobHolder.getJobKey());
                } catch (UnableToInterruptJobException e) {
                    log.error("Can't interrupt job {} for higher-priority job {} that will have to wait", runningJobHolder.activeJob, newJob);
                }
            }
        }
        return VETO;
    }

    /**
     * The interrupt() method of a InterruptableJob should issue a thread.interrupt() on the job thread,
     * and if not handled already, the resulting InterruptedException will be handled here.
     */
    @Override
    public void jobWasExecuted(JobExecutionContext context, JobExecutionException jobException) {
        // Just in case a job throws InterruptedException when interrupted.
        // Raise a flag if the job was interrupted with InterruptedException
        if (jobException!=null && jobException.getCause() instanceof InterruptedException) {
            PreemptiveVolatileQueueQuartzListener.setInterrupted(context);
        }
    }

    // Debug method
    private void printQueue(RunningJobHolder runningJobHolder) {
        if (log.isDebugEnabled()) {
            PriorityQueue<PrioritizedJob> clone = new PriorityQueue<PrioritizedJob>();
            clone.addAll(runningJobHolder.jobQueue);
            log.debug("Priority Queue: {}", clone.isEmpty()?"empty":"");
            while (!clone.isEmpty()) {
                PrioritizedJob job = clone.poll();
                String jobKey = (String) job.trigger.getJobDataMap().getOrDefault(PreemptiveVolatileQueueQuartzListener.JOB_ORIG_KEY, job.trigger.getJobKey().toString());
                log.debug("- {} [{}] reschedulings={}", job, jobKey, job.reschedulings);
            }
        }
    }

    // When a job finishes execution, a new one is started if the queue is not empty
    private boolean startNextJobInQueue(PrioritizedJob terminatedJob, RunningJobHolder runningJobHolder, String preemptedGroup, Trigger usedTrigger, JobExecutionContext context) {
        PrioritizedJob queuedJob = runningJobHolder.jobQueue.poll(); // Remove from queue
        if (queuedJob!=null) {          //
            log.debug("Starting next job in queue {} after job {} finished", queuedJob, terminatedJob);
            // The job must be cloned with a new trigger to execute immediately.
            // Can't reuse the existing jobDetail with a new trigger because for some reason when the original trigger has
            // finished all invocations, the following exception is thrown when trying to start the new trigger:
            // org.quartz.JobPersistenceException: The job (xxx) referenced by the trigger does not exist. 
            JobDataMap jobDataMap = queuedJob.jobDetail.getJobDataMap();
            JobDataMap triggerJobDataMap = queuedJob.trigger.getJobDataMap();
            JobDataMap newTriggerJobDataMap = new JobDataMap(triggerJobDataMap);
            // Need to store the original jobKey, used to check if a starting job is already in the queue. I can't use the normal
            // jobKey because, when a job is cloned here, its key must be changed in order to store it without a "job already exists" exception.
            String jobOrigKey = (String) triggerJobDataMap.getOrDefault(JOB_ORIG_KEY, queuedJob.jobDetail.getKey().toString());
            newTriggerJobDataMap.put(JOB_ORIG_KEY, jobOrigKey);
            JobDetail newJob = JobBuilder.newJob(queuedJob.jobDetail.getJobClass())
                .withIdentity(makePreemptedId(queuedJob.jobDetail.getKey().getName()), queuedJob.jobDetail.getKey().getGroup())
                .requestRecovery(queuedJob.jobDetail.requestsRecovery())
                .storeDurably(queuedJob.jobDetail.isDurable())
                .withDescription(queuedJob.jobDetail.getDescription())
                .usingJobData(jobDataMap)
                .build();
            Trigger newTrigger = newTrigger()
                .withPriority(queuedJob.priority)
                .withIdentity(makePreemptedId(queuedJob.trigger.getKey().getName()), preemptedGroup)
                .usingJobData(newTriggerJobDataMap)
                .withDescription(queuedJob.trigger.getDescription())
                .startNow()
                .withSchedule(simpleSchedule()
                    // A misfire occurs if a persistent trigger “misses” its firing time because 
                    // of the scheduler being shutdown, or because there are no available threads
                    .withMisfireHandlingInstructionFireNow() // (Not sure is correct)
                    )            
                .build();
            try {
                context.getScheduler().scheduleJob(newJob, newTrigger);
                log.debug("Job {} from queue rescheduled to start now as {}", queuedJob, makeJobString(newTrigger));
                queuedJob.reschedulings++;
                queuedJob.trigger = newTrigger;
                queuedJob.jobDetail = newJob;
                runningJobHolder.activeJob = queuedJob;
                return true;
            } catch (SchedulerException e) {
                log.error("Failed to start queued job {}", queuedJob, e);
                runningJobHolder.activeJob = null;
                return false;
            }
        }       
        return false;
    }

    private String makeJobString(Trigger trigger) {
        StringBuilder sb = new StringBuilder(trigger.getKey().toString());
        sb.append("-").append(trigger.getJobKey().toString());
        return sb.toString();
    }

    // Each time a job is rescheduled with a new trigger, their names are changed to a (hopefully) unique string
    private String makePreemptedId(String oldName) {
        final String marker = "_p_r_e_";
        long random = ThreadLocalRandom.current().nextLong(999888777L);
        StringBuffer result = new StringBuffer(Long.toString(random)); // nnn
        int pos = oldName.indexOf(marker);
        if (pos>-1) {
            result.append(oldName.substring(pos));
        } else {
            result.append(marker).append(oldName);
        }
        return result.toString();
    }

    // Called when a job finishes execution
    @Override
    public void triggerComplete(Trigger usedTrigger, JobExecutionContext context, CompletedExecutionInstruction completedExecutionInstruction) {
        boolean interruptedJob = isInterrupted(context);
        if (log.isDebugEnabled()) {
            if (interruptedJob) {
                log.debug("Interrupted job {}", makeJobString(usedTrigger));
            } else {
                log.debug("Terminated job {}", makeJobString(usedTrigger));
            }
        }
        String preemptedGroup = usedTrigger.getKey().getGroup();
        synchronized (runningJobs) {
            RunningJobHolder runningJobHolder = runningJobs.get(preemptedGroup);
            // Check that the activeJob is also the one that just terminated - for consistency
            if (runningJobHolder==null || !runningJobHolder.getJobKey().equals(context.getJobDetail().getKey())) {
                // Should never happen if there aren't any bugs
                log.error("Internal Error: the job in triggerComplete {} is not the active job {} for group {} (skipping)",
                    makeJobString(usedTrigger),
                    runningJobHolder==null?null:runningJobHolder.activeJob,
                    preemptedGroup);
                return;
            }
            printQueue(runningJobHolder);
            PrioritizedJob terminatedJob = runningJobHolder.activeJob;
            clearInterrupted(context);
            runningJobHolder.activeJob = null;
            // Start the next queued job if any. Do it in a loop because the next job might not start 
            // properly and this would otherwise prevent the other queued jobs from starting. 
            boolean started = false;
            while (!started && !runningJobHolder.jobQueue.isEmpty()) {
                started = startNextJobInQueue(terminatedJob, runningJobHolder, preemptedGroup, usedTrigger, context);
                if (interruptedJob && (started || runningJobHolder.jobQueue.isEmpty())) {
                    // It was an interrupted lower-priority job, so put it in the queue 
                    log.debug("Interrupted job {} added to job queue for rescheduling", terminatedJob);
                    runningJobHolder.addToQueue(terminatedJob);
                    interruptedJob=false;
                }
            }
            printQueue(runningJobHolder);
            if (runningJobHolder.jobQueue.isEmpty() && runningJobHolder.activeJob == null) {
                // The current terminated job was the last one in the trigger group, so we can clean up
                log.debug("Job {} ended with an empty proprity queue", terminatedJob);
                runningJobs.remove(preemptedGroup);
            }
        }
    }

    @Override
    public void triggerFired(Trigger trigger, JobExecutionContext context) {
    }

    @Override
    public void triggerMisfired(Trigger trigger) {
    }

    @Override
    public String getName() {
        return this.getClass().getSimpleName();
    }

    @Override
    public void jobToBeExecuted(JobExecutionContext context) {
    }

    @Override
    public void jobExecutionVetoed(JobExecutionContext context) {
    }
}


//
// RELATED CLASSES
//


/**
 * A job with associated priority and timestamp
 *
 */
class PrioritizedJob implements Comparable<PrioritizedJob> {
    int priority;
    long creationTimestamp = System.currentTimeMillis(); // To prevent same-priority elements to age
    Trigger trigger;
    JobDetail jobDetail; // Needed to create a new job when rescheduling after pulling from the queue
    int reschedulings = 0; // Number of times the job has been put back in the queue because preempted by a higher-priority job

    @Override
    public int compareTo(PrioritizedJob o) {
        // Smallest PrioritizedJob goes first, so priority check must be inverted because higher priority goes first
        int comparison = -(new Integer(this.priority).compareTo(o.priority));
        if (comparison==0) {
            // lower timestamp is higher priority
            comparison = new Long(this.creationTimestamp).compareTo(o.creationTimestamp);
        }
        return comparison;
    }

    @Override
    public String toString() {
        StringBuffer result = new StringBuffer();
        result.append(trigger.getKey()).append("-").append(trigger.getJobKey()).append("(").append(priority).append(")");
        return result.toString();
    }
}


/**
 * Holds the current running job definition for a given trigger group
 */
class RunningJobHolder {
    PrioritizedJob activeJob; // The running  job
    // This queue holds all jobs of the same thread group that tried to start while this job was running.
    // They are sorted by priority and timestamp.
    // The head of the queue might contain a higher-priority job that has been put in the queue while waiting for
    // the active job to handle the interruption
    PriorityQueue<PrioritizedJob> jobQueue = new PriorityQueue<>(); 

    JobKey getJobKey() {
        return activeJob.trigger.getJobKey();
    }

    /**
     * Create a new PrioritizedJob and set it as active
     * @param startingJobPriority
     * @param startingTrigger
     * @param startingJobContext
     * @return the new PrioritizedJob
     */
    PrioritizedJob setActiveJob(int startingJobPriority, Trigger startingTrigger, JobExecutionContext startingJobContext) {
        PrioritizedJob newJob = new PrioritizedJob();
        newJob.priority = startingJobPriority;
        newJob.trigger = startingTrigger;
        newJob.jobDetail = startingJobContext.getJobDetail();
        this.activeJob = newJob;
        return newJob;
    }

    /**
     * Create a new PrioritizedJob and add it to the queue
     * @param startingJobPriority
     * @param startingTrigger
     * @param startingJobContext
     * @return the new PrioritizedJob
     */
    PrioritizedJob queueJob(int startingJobPriority, Trigger startingTrigger, JobExecutionContext startingJobContext) {
        PrioritizedJob newJob = new PrioritizedJob();
        newJob.priority = startingJobPriority;
        newJob.trigger = startingTrigger;
        newJob.jobDetail = startingJobContext.getJobDetail();
        addToQueue(newJob);
        return newJob;
    }

    /**
     * Compares job keys, first by fetching the original job key stored in the trigger JobDataMap, then by using the job's own key
     * @param trigger
     * @param prioritizedJob
     * @return
     */
    private boolean equalKeys(Trigger trigger, PrioritizedJob prioritizedJob) {
        String triggerJobKeyToCheck = (String) trigger.getJobDataMap().getOrDefault(PreemptiveVolatileQueueQuartzListener.JOB_ORIG_KEY, trigger.getJobKey().toString()); 
        String prioritizedJobKeyToCheck = (String) prioritizedJob.trigger.getJobDataMap().getOrDefault(PreemptiveVolatileQueueQuartzListener.JOB_ORIG_KEY, prioritizedJob.trigger.getJobKey().toString());
        return triggerJobKeyToCheck.equals(prioritizedJobKeyToCheck);
    }

    /**
     * Check if the job in a trigger has already been queued (or is the active one) by comparing the job key
     * @param trigger
     * @return
     */
    boolean isInHolder(Trigger trigger) {
        if (equalKeys(trigger, activeJob)) {
            return true;
        }
        for (PrioritizedJob prioritizedJob : jobQueue) {
            if (equalKeys(trigger, prioritizedJob)) {
                return true;
            }
        }
        return false;
    }

    void addToQueue(PrioritizedJob prioritizedJob) {
        jobQueue.add(prioritizedJob);
    }
}