org.quartz.spi.OperableTrigger#getJobKey ( )源码实例Demo

/**
 * Populate entity fields used for indexing.
 */
@Override
public void setValue(final OperableTrigger value) {
  super.setValue(value);

  TriggerKey key = value.getKey();
  checkState(key != null, "Missing key");
  setName(key.getName());
  setGroup(key.getGroup());

  JobKey jobKey = value.getJobKey();
  checkState(jobKey != null, "Missing job-key");
  setJobName(jobKey.getName());
  setJobGroup(jobKey.getGroup());

  setCalendarName(value.getCalendarName());
}
 

TriggerWrapper(OperableTrigger trigger) {
    if(trigger == null)
        throw new IllegalArgumentException("Trigger cannot be null!");
    this.trigger = trigger;
    key = trigger.getKey();
    this.jobKey = trigger.getJobKey();
}
 

/**
 * <p>
 * Store the given <code>{@link org.quartz.Trigger}</code>.
 * </p>
 * 
 * @param newTrigger The <code>Trigger</code> to be stored.
 * @param replaceExisting If <code>true</code>, any <code>Trigger</code> existing in the <code>JobStore</code> with
 *        the same name & group should be over-written.
 * @throws ObjectAlreadyExistsException if a <code>Trigger</code> with the same name/group already exists, and
 *         replaceExisting is set to false.
 * @see #pauseTriggers(org.quartz.impl.matchers.GroupMatcher)
 */
@Override
public void storeTrigger(OperableTrigger newTrigger, boolean replaceExisting) throws JobPersistenceException {
  OperableTrigger clone = (OperableTrigger) newTrigger.clone();

  lock();
  try {
    JobDetail job = retrieveJob(newTrigger.getJobKey());
    if (job == null) {
      //
      throw new JobPersistenceException("The job (" + newTrigger.getJobKey()
                                        + ") referenced by the trigger does not exist.");
    }

    // wrapper construction must be done in lock since serializer is unlocked
    TriggerWrapper tw = wrapperFactory.createTriggerWrapper(clone, job.isConcurrentExectionDisallowed());

    if (triggerFacade.containsKey(tw.getKey())) {
      if (!replaceExisting) { throw new ObjectAlreadyExistsException(newTrigger); }

      removeTrigger(newTrigger.getKey(), false);
    }

    // add to triggers by group
    Set<String> grpSet = toolkitDSHolder.getOrCreateTriggersGroupMap(newTrigger.getKey().getGroup());
    grpSet.add(newTrigger.getKey().getName());
    if (!triggerFacade.hasGroup(newTrigger.getKey().getGroup())) {
      triggerFacade.addGroup(newTrigger.getKey().getGroup());
    }

    if (triggerFacade.pausedGroupsContain(newTrigger.getKey().getGroup())
        || jobFacade.pausedGroupsContain(newTrigger.getJobKey().getGroup())) {
      tw.setState(TriggerState.PAUSED, terracottaClientId, triggerFacade);
      if (jobFacade.blockedJobsContain(tw.getJobKey())) {
        tw.setState(TriggerState.PAUSED_BLOCKED, terracottaClientId, triggerFacade);
      }
    } else if (jobFacade.blockedJobsContain(tw.getJobKey())) {
      tw.setState(TriggerState.BLOCKED, terracottaClientId, triggerFacade);
    } else {
      timeTriggers.add(tw);
    }

    // add to triggers by FQN map
    triggerFacade.put(tw.getKey(), tw);
  } finally {
    unlock();
  }
}
 

/**
 * <p>
 * Store the given <code>{@link org.quartz.Trigger}</code>.
 * </p>
 *
 * @param newTrigger
 *          The <code>Trigger</code> to be stored.
 * @param replaceExisting
 *          If <code>true</code>, any <code>Trigger</code> existing in
 *          the <code>JobStore</code> with the same name & group should
 *          be over-written.
 * @throws ObjectAlreadyExistsException
 *           if a <code>Trigger</code> with the same name/group already
 *           exists, and replaceExisting is set to false.
 *
 * @see #pauseTriggers(org.quartz.impl.matchers.GroupMatcher)
 */
public void storeTrigger(OperableTrigger newTrigger,
        boolean replaceExisting) throws JobPersistenceException {
    TriggerWrapper tw = new TriggerWrapper((OperableTrigger)newTrigger.clone());

    synchronized (lock) {
        if (triggersByKey.get(tw.key) != null) {
            if (!replaceExisting) {
                throw new ObjectAlreadyExistsException(newTrigger);
            }

            removeTrigger(newTrigger.getKey(), false);
        }

        if (retrieveJob(newTrigger.getJobKey()) == null) {
            throw new JobPersistenceException("The job ("
                    + newTrigger.getJobKey()
                    + ") referenced by the trigger does not exist.");
        }

        // add to triggers array
        triggers.add(tw);
        // add to triggers by group
        HashMap<TriggerKey, TriggerWrapper> grpMap = triggersByGroup.get(newTrigger.getKey().getGroup());
        if (grpMap == null) {
            grpMap = new HashMap<TriggerKey, TriggerWrapper>(100);
            triggersByGroup.put(newTrigger.getKey().getGroup(), grpMap);
        }
        grpMap.put(newTrigger.getKey(), tw);
        // add to triggers by FQN map
        triggersByKey.put(tw.key, tw);

        if (pausedTriggerGroups.contains(newTrigger.getKey().getGroup())
                || pausedJobGroups.contains(newTrigger.getJobKey().getGroup())) {
            tw.state = TriggerWrapper.STATE_PAUSED;
            if (blockedJobs.contains(tw.jobKey)) {
                tw.state = TriggerWrapper.STATE_PAUSED_BLOCKED;
            }
        } else if (blockedJobs.contains(tw.jobKey)) {
            tw.state = TriggerWrapper.STATE_BLOCKED;
        } else {
            timeTriggers.add(tw);
        }
    }
}
 

/**
 * <p>
 * Insert or update a trigger.
 * </p>
 */
@SuppressWarnings("ConstantConditions")
protected void storeTrigger(Connection conn,
        OperableTrigger newTrigger, JobDetail job, boolean replaceExisting, String state,
        boolean forceState, boolean recovering)
    throws JobPersistenceException {

    boolean existingTrigger = triggerExists(conn, newTrigger.getKey());

    if ((existingTrigger) && (!replaceExisting)) { 
        throw new ObjectAlreadyExistsException(newTrigger); 
    }
    
    try {

        boolean shouldBepaused;

        if (!forceState) {
            shouldBepaused = getDelegate().isTriggerGroupPaused(
                    conn, newTrigger.getKey().getGroup());

            if(!shouldBepaused) {
                shouldBepaused = getDelegate().isTriggerGroupPaused(conn,
                        ALL_GROUPS_PAUSED);

                if (shouldBepaused) {
                    getDelegate().insertPausedTriggerGroup(conn, newTrigger.getKey().getGroup());
                }
            }

            if (shouldBepaused && (state.equals(STATE_WAITING) || state.equals(STATE_ACQUIRED))) {
                state = STATE_PAUSED;
            }
        }

        if(job == null) {
            job = retrieveJob(conn, newTrigger.getJobKey());
        }
        if (job == null) {
            throw new JobPersistenceException("The job ("
                    + newTrigger.getJobKey()
                    + ") referenced by the trigger does not exist.");
        }

        if (job.isConcurrentExectionDisallowed() && !recovering) { 
            state = checkBlockedState(conn, job.getKey(), state);
        }
        
        if (existingTrigger) {
            getDelegate().updateTrigger(conn, newTrigger, state, job);
        } else {
            getDelegate().insertTrigger(conn, newTrigger, state, job);
        }
    } catch (Exception e) {
        throw new JobPersistenceException("Couldn't store trigger '" + newTrigger.getKey() + "' for '" 
                + newTrigger.getJobKey() + "' job:" + e.getMessage(), e);
    }
}
 

protected List<OperableTrigger> acquireNextTrigger(Connection conn, long noLaterThan, int maxCount, long timeWindow)
    throws JobPersistenceException {
    if (timeWindow < 0) {
      throw new IllegalArgumentException();
    }
    
    List<OperableTrigger> acquiredTriggers = new ArrayList<OperableTrigger>();
    Set<JobKey> acquiredJobKeysForNoConcurrentExec = new HashSet<JobKey>();
    final int MAX_DO_LOOP_RETRY = 3;
    int currentLoopCount = 0;
    do {
        currentLoopCount ++;
        try {
            List<TriggerKey> keys = getDelegate().selectTriggerToAcquire(conn, noLaterThan + timeWindow, getMisfireTime(), maxCount);
            
            // No trigger is ready to fire yet.
            if (keys == null || keys.size() == 0)
                return acquiredTriggers;

            long batchEnd = noLaterThan;

            for(TriggerKey triggerKey: keys) {
                // If our trigger is no longer available, try a new one.
                OperableTrigger nextTrigger = retrieveTrigger(conn, triggerKey);
                if(nextTrigger == null) {
                    continue; // next trigger
                }
                
                // If trigger's job is set as @DisallowConcurrentExecution, and it has already been added to result, then
                // put it back into the timeTriggers set and continue to search for next trigger.
                JobKey jobKey = nextTrigger.getJobKey();
                JobDetail job;
                try {
                    job = retrieveJob(conn, jobKey);
                } catch (JobPersistenceException jpe) {
                    try {
                        getLog().error("Error retrieving job, setting trigger state to ERROR.", jpe);
                        getDelegate().updateTriggerState(conn, triggerKey, STATE_ERROR);
                    } catch (SQLException sqle) {
                        getLog().error("Unable to set trigger state to ERROR.", sqle);
                    }
                    continue;
                }
                
                if (job.isConcurrentExectionDisallowed()) {
                    if (acquiredJobKeysForNoConcurrentExec.contains(jobKey)) {
                        continue; // next trigger
                    } else {
                        acquiredJobKeysForNoConcurrentExec.add(jobKey);
                    }
                }
                
                if (nextTrigger.getNextFireTime().getTime() > batchEnd) {
                  break;
                }
                // We now have a acquired trigger, let's add to return list.
                // If our trigger was no longer in the expected state, try a new one.
                int rowsUpdated = getDelegate().updateTriggerStateFromOtherState(conn, triggerKey, STATE_ACQUIRED, STATE_WAITING);
                if (rowsUpdated <= 0) {
                    continue; // next trigger
                }
                nextTrigger.setFireInstanceId(getFiredTriggerRecordId());
                getDelegate().insertFiredTrigger(conn, nextTrigger, STATE_ACQUIRED, null);

                if(acquiredTriggers.isEmpty()) {
                    batchEnd = Math.max(nextTrigger.getNextFireTime().getTime(), System.currentTimeMillis()) + timeWindow;
                }
                acquiredTriggers.add(nextTrigger);
            }

            // if we didn't end up with any trigger to fire from that first
            // batch, try again for another batch. We allow with a max retry count.
            if(acquiredTriggers.size() == 0 && currentLoopCount < MAX_DO_LOOP_RETRY) {
                continue;
            }
            
            // We are done with the while loop.
            break;
        } catch (Exception e) {
            throw new JobPersistenceException(
                      "Couldn't acquire next trigger: " + e.getMessage(), e);
        }
    } while (true);
    
    // Return the acquired trigger list
    return acquiredTriggers;
}