聊聊artemis的scheduledDeliveryTime

序

本文主要研究一下artemis的scheduledDeliveryTime

HDR_SCHEDULED_DELIVERY_TIME

activemq-artemis-2.11.0/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/CoreMessage.java

public class CoreMessage extends RefCountMessage implements ICoreMessage {

   //......

   @Override
   public Long getScheduledDeliveryTime() {
      Object property = getProperties().getProperty(Message.HDR_SCHEDULED_DELIVERY_TIME);

      if (property != null && property instanceof Number) {
         return ((Number) property).longValue();
      }

      return 0L;
   }

   @Override
   public CoreMessage setScheduledDeliveryTime(Long time) {
      if (time == null || time == 0) {
         getProperties().removeProperty(Message.HDR_SCHEDULED_DELIVERY_TIME);
      } else {
         putLongProperty(Message.HDR_SCHEDULED_DELIVERY_TIME, time);
      }
      return this;
   }

   //......
}

• CoreMessage的getScheduledDeliveryTime會讀取Message.HDR_SCHEDULED_DELIVERY_TIME屬性的值；setScheduledDeliveryTime方法則會更新Message.HDR_SCHEDULED_DELIVERY_TIME屬性的值

processRoute

activemq-artemis-2.11.0/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java

public class PostOfficeImpl implements PostOffice, NotificationListener, BindingsFactory {

   //......

   public void processRoute(final Message message,
                            final RoutingContext context,
                            final boolean direct) throws Exception {
      final List<MessageReference> refs = new ArrayList<>();

      Transaction tx = context.getTransaction();

      Long deliveryTime = message.getScheduledDeliveryTime();

      for (Map.Entry<SimpleString, RouteContextList> entry : context.getContexListing().entrySet()) {
         PagingStore store = pagingManager.getPageStore(entry.getKey());

         if (store != null && storageManager.addToPage(store, message, context.getTransaction(), entry.getValue())) {
            if (message.isLargeMessage()) {
               confirmLargeMessageSend(tx, message);
            }

            // We need to kick delivery so the Queues may check for the cursors case they are empty
            schedulePageDelivery(tx, entry);
            continue;
         }

         for (Queue queue : entry.getValue().getNonDurableQueues()) {
            MessageReference reference = MessageReference.Factory.createReference(message, queue);

            if (deliveryTime != null) {
               reference.setScheduledDeliveryTime(deliveryTime);
            }
            refs.add(reference);

            message.incrementRefCount();
         }

         Iterator<Queue> iter = entry.getValue().getDurableQueues().iterator();

         while (iter.hasNext()) {
            Queue queue = iter.next();

            MessageReference reference = MessageReference.Factory.createReference(message, queue);

            if (context.isAlreadyAcked(context.getAddress(message), queue)) {
               reference.setAlreadyAcked();
               if (tx != null) {
                  queue.acknowledge(tx, reference);
               }
            }

            if (deliveryTime != null) {
               reference.setScheduledDeliveryTime(deliveryTime);
            }
            refs.add(reference);

            if (message.isDurable()) {
               int durableRefCount = message.incrementDurableRefCount();

               if (durableRefCount == 1) {
                  if (tx != null) {
                     storageManager.storeMessageTransactional(tx.getID(), message);
                  } else {
                     storageManager.storeMessage(message);
                  }

                  if (message.isLargeMessage()) {
                     confirmLargeMessageSend(tx, message);
                  }
               }

               if (tx != null) {
                  storageManager.storeReferenceTransactional(tx.getID(), queue.getID(), message.getMessageID());

                  tx.setContainsPersistent();
               } else {
                  storageManager.storeReference(queue.getID(), message.getMessageID(), !iter.hasNext());
               }

               if (deliveryTime != null && deliveryTime > 0) {
                  if (tx != null) {
                     storageManager.updateScheduledDeliveryTimeTransactional(tx.getID(), reference);
                  } else {
                     storageManager.updateScheduledDeliveryTime(reference);
                  }
               }
            }

            message.incrementRefCount();
         }
      }

      if (tx != null) {
         tx.addOperation(new AddOperation(refs));
      } else {
         // This will use the same thread if there are no pending operations
         // avoiding a context switch on this case
         storageManager.afterCompleteOperations(new IOCallback() {
            @Override
            public void onError(final int errorCode, final String errorMessage) {
               ActiveMQServerLogger.LOGGER.ioErrorAddingReferences(errorCode, errorMessage);
            }

            @Override
            public void done() {
               context.processReferences(refs, direct);
            }
         });
      }
   }

   //......

}

• PostOfficeImpl的processRoute方法會將message.getScheduledDeliveryTime()更新到reference.setScheduledDeliveryTime(deliveryTime)；對於isDurable爲true的message還會執行storageManager.storeMessage；對於tx不爲null的會建立AddOperation添加到tx中，對於tx爲null的則經過storageManager.afterCompleteOperations註冊IOCallback，其done方法執行context.processReferences(refs, direct)

checkAndSchedule

activemq-artemis-2.11.0/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerImpl.java

public class ScheduledDeliveryHandlerImpl implements ScheduledDeliveryHandler {

   private static final Logger logger = Logger.getLogger(ScheduledDeliveryHandlerImpl.class);

   private final ScheduledExecutorService scheduledExecutor;

   private final Map<Long, Runnable> runnables = new ConcurrentHashMap<>();

   //......

   public boolean checkAndSchedule(final MessageReference ref, final boolean tail) {
      long deliveryTime = ref.getScheduledDeliveryTime();

      if (deliveryTime > 0 && scheduledExecutor != null) {
         if (logger.isTraceEnabled()) {
            logger.trace("Scheduling delivery for " + ref + " to occur at " + deliveryTime);
         }

         addInPlace(deliveryTime, ref, tail);

         scheduleDelivery(deliveryTime);

         return true;
      }
      return false;
   }

   private void scheduleDelivery(final long deliveryTime) {
      final long now = System.currentTimeMillis();

      final long delay = deliveryTime - now;

      if (delay < 0) {
         if (logger.isTraceEnabled()) {
            logger.trace("calling another scheduler now as deliverTime " + deliveryTime + " < now=" + now);
         }
         // if delay == 0 we will avoid races between adding the scheduler and finishing it
         ScheduledDeliveryRunnable runnable = new ScheduledDeliveryRunnable(deliveryTime);
         scheduledExecutor.schedule(runnable, 0, TimeUnit.MILLISECONDS);
      } else if (!runnables.containsKey(deliveryTime)) {
         ScheduledDeliveryRunnable runnable = new ScheduledDeliveryRunnable(deliveryTime);

         if (logger.isTraceEnabled()) {
            logger.trace("Setting up scheduler for " + deliveryTime + " with a delay of " + delay + " as now=" + now);
         }

         runnables.put(deliveryTime, runnable);
         scheduledExecutor.schedule(runnable, delay, TimeUnit.MILLISECONDS);
      } else {
         if (logger.isTraceEnabled()) {
            logger.trace("Couldn't make another scheduler as " + deliveryTime + " is already set, now is " + now);
         }
      }
   }

   //......
}

• ScheduledDeliveryHandlerImpl的checkAndSchedule方法會執行scheduleDelivery，該方法對於delay小於0的會立馬往建立ScheduledDeliveryRunnable並提交scheduledExecutor調度；不然會維護deliveryTime以及ScheduledDeliveryRunnable的映射到runnables中，並往scheduledExecutor提交runnable延時執行

ScheduledDeliveryRunnable

activemq-artemis-2.11.0/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerImpl.java

private class ScheduledDeliveryRunnable implements Runnable {

      long deliveryTime;

      private ScheduledDeliveryRunnable(final long deliveryTime) {
         this.deliveryTime = deliveryTime;
      }

      @Override
      public void run() {
         HashMap<Queue, LinkedList<MessageReference>> refs = new HashMap<>();

         runnables.remove(deliveryTime);

         final long now = System.currentTimeMillis();

         if (now < deliveryTime) {
            // Ohhhh... blame it on the OS
            // on some OSes (so far Windows only) the precision of the scheduled executor could eventually give
            // an executor call earlier than it was supposed...
            // for that reason we will schedule it again so no messages are lost!
            // we can't just assume deliveryTime here as we could deliver earlier than what we are supposed to
            // this is basically a hack to work around an OS or JDK bug!
            if (logger.isTraceEnabled()) {
               logger.trace("Scheduler is working around OS imprecisions on " +
                               "timing and re-scheduling an executor. now=" + now +
                               " and deliveryTime=" + deliveryTime);
            }
            ScheduledDeliveryHandlerImpl.this.scheduleDelivery(deliveryTime);
         }

         if (logger.isTraceEnabled()) {
            logger.trace("Is it " + System.currentTimeMillis() + " now and we are running deliveryTime = " + deliveryTime);
         }

         synchronized (scheduledReferences) {

            Iterator<RefScheduled> iter = scheduledReferences.iterator();
            while (iter.hasNext()) {
               MessageReference reference = iter.next().getRef();
               if (reference.getScheduledDeliveryTime() > now) {
                  // We will delivery as long as there are messages to be delivered
                  break;
               }

               iter.remove();
               metrics.decrementMetrics(reference);

               reference.setScheduledDeliveryTime(0);

               LinkedList<MessageReference> references = refs.get(reference.getQueue());

               if (references == null) {
                  references = new LinkedList<>();
                  refs.put(reference.getQueue(), references);
               }

               if (logger.isTraceEnabled()) {
                  logger.trace("sending message " + reference + " to delivery, deliveryTime =  " + deliveryTime);
               }

               references.addFirst(reference);
            }
            if (logger.isTraceEnabled()) {
               logger.trace("Finished loop on deliveryTime = " + deliveryTime);
            }
         }

         for (Map.Entry<Queue, LinkedList<MessageReference>> entry : refs.entrySet()) {

            Queue queue = entry.getKey();
            LinkedList<MessageReference> list = entry.getValue();
            if (logger.isTraceEnabled()) {
               logger.trace("Delivering " + list.size() + " elements on list to queue " + queue);
            }
            queue.addHead(list, true);
         }

         // Just to speed up GC
         refs.clear();
      }
   }

• ScheduledDeliveryRunnable實現了Runnable方法，其構造器接收deliveryTime參數，其run方法會先移除指定deliveryTime的Runnable，對於當前時間小於deliveryTime的執行ScheduledDeliveryHandlerImpl.this.scheduleDelivery(deliveryTime)；以後遍歷scheduledReferences取出reference.getScheduledDeliveryTime()小於等於now的，執行reference.setScheduledDeliveryTime(0)，而後添加到refs；以後挨個遍歷refs，執行queue.addHead(list, true)(其中list爲LinkedList<MessageReference>)

addTail

afterCommit

activemq-artemis-2.11.0/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java

public static final class AddOperation implements TransactionOperation {

      private final List<MessageReference> refs;

      AddOperation(final List<MessageReference> refs) {
         this.refs = refs;
      }

      @Override
      public void afterCommit(final Transaction tx) {
         for (MessageReference ref : refs) {
            if (!ref.isAlreadyAcked()) {
               ref.getQueue().addTail(ref, false);
            }
         }
      }

      @Override
      public void afterPrepare(final Transaction tx) {
         for (MessageReference ref : refs) {
            if (ref.isAlreadyAcked()) {
               ref.getQueue().referenceHandled(ref);
               ref.getQueue().incrementMesssagesAdded();
            }
         }
      }

      //......

   }

• AddOperation的afterCommit方法對於非isAlreadyAcked的會執行ref.getQueue().addTail(ref, false)

processReferences

activemq-artemis-2.11.0/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RoutingContextImpl.java

public final class RoutingContextImpl implements RoutingContext {

   //......

   public void processReferences(final List<MessageReference> refs, final boolean direct) {
      internalprocessReferences(refs, direct);
   }

   private void internalprocessReferences(final List<MessageReference> refs, final boolean direct) {
      for (MessageReference ref : refs) {
         ref.getQueue().addTail(ref, direct);
      }
   }

   //......
}

• RoutingContextImpl的processReferences方法執行ref.getQueue().addTail(ref, direct)

addTail

activemq-artemis-2.11.0/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java

public class QueueImpl extends CriticalComponentImpl implements Queue {

   //......

   private final MpscUnboundedArrayQueue<MessageReference> intermediateMessageReferences = new MpscUnboundedArrayQueue<>(8192);

   //......   

   public void addTail(final MessageReference ref, final boolean direct) {
      enterCritical(CRITICAL_PATH_ADD_TAIL);
      try {
         if (scheduleIfPossible(ref)) {
            return;
         }

         if (direct && supportsDirectDeliver && !directDeliver && System.currentTimeMillis() - lastDirectDeliveryCheck > CHECK_QUEUE_SIZE_PERIOD) {
            if (logger.isTraceEnabled()) {
               logger.trace("Checking to re-enable direct deliver on queue " + this.getName());
            }
            lastDirectDeliveryCheck = System.currentTimeMillis();
            synchronized (directDeliveryGuard) {
               // The checkDirect flag is periodically set to true, if the delivery is specified as direct then this causes the
               // directDeliver flag to be re-computed resulting in direct delivery if the queue is empty
               // We don't recompute it on every delivery since executing isEmpty is expensive for a ConcurrentQueue

               if (deliveriesInTransit.getCount() == 0 && getExecutor().isFlushed() &&
                  intermediateMessageReferences.isEmpty() && messageReferences.isEmpty() &&
                  pageIterator != null && !pageIterator.hasNext() &&
                  pageSubscription != null && !pageSubscription.isPaging()) {
                  // We must block on the executor to ensure any async deliveries have completed or we might get out of order
                  // deliveries
                  // Go into direct delivery mode
                  directDeliver = supportsDirectDeliver;
                  if (logger.isTraceEnabled()) {
                     logger.trace("Setting direct deliverer to " + supportsDirectDeliver + " on queue " + this.getName());
                  }
               } else {
                  if (logger.isTraceEnabled()) {
                     logger.trace("Couldn't set direct deliver back on queue " + this.getName());
                  }
               }
            }
         }

         if (direct && supportsDirectDeliver && directDeliver && deliveriesInTransit.getCount() == 0 && deliverDirect(ref)) {
            return;
         }

         // We only add queueMemorySize if not being delivered directly
         queueMemorySize.addAndGet(ref.getMessageMemoryEstimate());

         intermediateMessageReferences.add(ref);

         directDeliver = false;

         // Delivery async will both poll for intermediate reference and deliver to clients
         deliverAsync();
      } finally {
         leaveCritical(CRITICAL_PATH_ADD_TAIL);
      }
   }

   protected boolean scheduleIfPossible(MessageReference ref) {
      if (scheduledDeliveryHandler.checkAndSchedule(ref, true)) {
         synchronized (this) {
            if (!ref.isPaged()) {
               incrementMesssagesAdded();
            }
         }

         return true;
      }
      return false;
   }

   //......
}

• QueueImpl的addTail方法會先執行scheduleIfPossible，若爲true則提早返回；若爲false的話則繼續往下，將MessageReference添加到intermediateMessageReferences；scheduleIfPossible方法則執行scheduledDeliveryHandler.checkAndSchedule(ref, true)，其返回true則爲true，不然爲false

doInternalPoll

activemq-artemis-2.11.0/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java

public class QueueImpl extends CriticalComponentImpl implements Queue {

   //......

   private synchronized void doInternalPoll() {

      int added = 0;
      MessageReference ref;

      while ((ref = intermediateMessageReferences.poll()) != null) {
         internalAddTail(ref);

         if (!ref.isPaged()) {
            incrementMesssagesAdded();
         }

         if (added++ > MAX_DELIVERIES_IN_LOOP) {
            // if we just keep polling from the intermediate we could starve in case there's a sustained load
            deliverAsync(true);
            return;
         }
      }
   }

   private synchronized void internalAddTail(final MessageReference ref) {
      refAdded(ref);
      messageReferences.addTail(ref, getPriority(ref));
      pendingMetrics.incrementMetrics(ref);
      enforceRing(false);
   }

   //......
}

• QueueImpl的doInternalPoll方法會將intermediateMessageReferences裏頭的message轉移到messageReferences中

addHead

handleAddMessage

activemq-artemis-2.11.0/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/PostOfficeJournalLoader.java

public class PostOfficeJournalLoader implements JournalLoader {

   //......

   public void handleAddMessage(Map<Long, Map<Long, AddMessageRecord>> queueMap) throws Exception {
      for (Map.Entry<Long, Map<Long, AddMessageRecord>> entry : queueMap.entrySet()) {
         long queueID = entry.getKey();

         Map<Long, AddMessageRecord> queueRecords = entry.getValue();

         Queue queue = this.queues.get(queueID);

         if (queue == null) {
            if (queueRecords.values().size() != 0) {
               ActiveMQServerLogger.LOGGER.journalCannotFindQueueForMessage(queueID);
            }

            continue;
         }

         // Redistribution could install a Redistributor while we are still loading records, what will be an issue with
         // prepared ACKs
         // We make sure te Queue is paused before we reroute values.
         queue.pause();

         Collection<AddMessageRecord> valueRecords = queueRecords.values();

         long currentTime = System.currentTimeMillis();

         for (AddMessageRecord record : valueRecords) {
            long scheduledDeliveryTime = record.getScheduledDeliveryTime();

            if (scheduledDeliveryTime != 0 && scheduledDeliveryTime <= currentTime) {
               scheduledDeliveryTime = 0;
               record.getMessage().setScheduledDeliveryTime(0L);
            }

            if (scheduledDeliveryTime != 0) {
               record.getMessage().setScheduledDeliveryTime(scheduledDeliveryTime);
            }

            MessageReference ref = postOffice.reroute(record.getMessage(), queue, null);

            ref.setDeliveryCount(record.getDeliveryCount());

            if (scheduledDeliveryTime != 0) {
               record.getMessage().setScheduledDeliveryTime(0L);
            }
         }
      }
   }

   //......
}

• PostOfficeJournalLoader的handleAddMessage會執行postOffice.reroute

reroute

workspace/activemq-artemis-2.11.0/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java

public class PostOfficeImpl implements PostOffice, NotificationListener, BindingsFactory {

   //......

   public MessageReference reroute(final Message message, final Queue queue, final Transaction tx) throws Exception {

      setPagingStore(queue.getAddress(), message);

      MessageReference reference = MessageReference.Factory.createReference(message, queue);

      Long scheduledDeliveryTime = message.getScheduledDeliveryTime();
      if (scheduledDeliveryTime != null) {
         reference.setScheduledDeliveryTime(scheduledDeliveryTime);
      }

      message.incrementDurableRefCount();

      message.incrementRefCount();

      if (tx == null) {
         queue.reload(reference);
      } else {
         List<MessageReference> refs = new ArrayList<>(1);

         refs.add(reference);

         tx.addOperation(new AddOperation(refs));
      }

      return reference;
   }

   //......
}

• PostOfficeImpl的reroute方法對於tx爲null執行queue.reload(reference)，對於tx不爲null的則建立AddOperation添加到tx

reload

activemq-artemis-2.11.0/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java

public class QueueImpl extends CriticalComponentImpl implements Queue {

   //......

   public synchronized void reload(final MessageReference ref) {
      queueMemorySize.addAndGet(ref.getMessageMemoryEstimate());
      if (!scheduledDeliveryHandler.checkAndSchedule(ref, true)) {
         internalAddTail(ref);
      }

      directDeliver = false;

      if (!ref.isPaged()) {
         incrementMesssagesAdded();
      }
   }

   private synchronized void internalAddTail(final MessageReference ref) {
      refAdded(ref);
      messageReferences.addTail(ref, getPriority(ref));
      pendingMetrics.incrementMetrics(ref);
      enforceRing(false);
   }

   //......
}

• QueueImpl的reload會先執行scheduledDeliveryHandler.checkAndSchedule(ref, true)，若返回false則執行internalAddTail(ref)；internalAddTail則執行messageReferences.addTail

scaleDownMessages

activemq-artemis-2.11.0/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ScaleDownHandler.java

public class ScaleDownHandler {

   //......

   public long scaleDownMessages(ClientSessionFactory sessionFactory,
                                 SimpleString nodeId,
                                 String user,
                                 String password) throws Exception {
      long messageCount = 0;
      targetNodeId = nodeId != null ? nodeId.toString() : getTargetNodeId(sessionFactory);

      try (ClientSession session = sessionFactory.createSession(user, password, false, true, true, false, 0)) {
         ClientProducer producer = session.createProducer();

         // perform a loop per address
         for (SimpleString address : postOffice.getAddresses()) {
            logger.debug("Scaling down address " + address);
            Bindings bindings = postOffice.lookupBindingsForAddress(address);

            // It will get a list of queues on this address, ordered by the number of messages
            Set<Queue> queues = new TreeSet<>(new OrderQueueByNumberOfReferencesComparator());
            if (bindings != null) {
               for (Binding binding : bindings.getBindings()) {
                  if (binding instanceof LocalQueueBinding) {
                     Queue queue = ((LocalQueueBinding) binding).getQueue();
                     // as part of scale down we will cancel any scheduled message and pass it to theWhile we scan for the queues we will also cancel any scheduled messages and deliver them right away
                     queue.deliverScheduledMessages();
                     queues.add(queue);
                  }
               }
            }

            String sfPrefix =  ((PostOfficeImpl) postOffice).getServer().getInternalNamingPrefix() + "sf.";
            if (address.toString().startsWith(sfPrefix)) {
               messageCount += scaleDownSNF(address, queues, producer);
            } else {
               messageCount += scaleDownRegularMessages(address, queues, session, producer);
            }

         }
      }

      return messageCount;
   }

   //......
}

• ScaleDownHandler的scaleDownMessages方法會執行queue.deliverScheduledMessages()

deliverScheduledMessages

activemq-artemis-2.11.0/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java

public class QueueImpl extends CriticalComponentImpl implements Queue {

   //......

   private final PriorityLinkedList<MessageReference> messageReferences = new PriorityLinkedListImpl<>(QueueImpl.NUM_PRIORITIES, MessageReferenceImpl.getIDComparator());

   //......

   public void deliverScheduledMessages() throws ActiveMQException {
      List<MessageReference> scheduledMessages = scheduledDeliveryHandler.cancel(null);
      if (scheduledMessages != null && scheduledMessages.size() > 0) {
         for (MessageReference ref : scheduledMessages) {
            ref.getMessage().setScheduledDeliveryTime(ref.getScheduledDeliveryTime());
            ref.setScheduledDeliveryTime(0);
         }
         this.addHead(scheduledMessages, true);
      }
   }

   public void addHead(final List<MessageReference> refs, boolean scheduling) {
      enterCritical(CRITICAL_PATH_ADD_HEAD);
      synchronized (this) {
         try {
            for (MessageReference ref : refs) {
               addHead(ref, scheduling);
            }

            resetAllIterators();

            deliverAsync();
         } finally {
            leaveCritical(CRITICAL_PATH_ADD_HEAD);
         }
      }
   }

   public void addHead(final MessageReference ref, boolean scheduling) {
      enterCritical(CRITICAL_PATH_ADD_HEAD);
      synchronized (this) {
         try {
            if (ringSize != -1) {
               enforceRing(ref, scheduling, true);
            }

            if (!ref.isAlreadyAcked()) {
               if (!scheduling && scheduledDeliveryHandler.checkAndSchedule(ref, false)) {
                  return;
               }

               internalAddHead(ref);

               directDeliver = false;
            }
         } finally {
            leaveCritical(CRITICAL_PATH_ADD_HEAD);
         }
      }
   }

   private void internalAddHead(final MessageReference ref) {
      queueMemorySize.addAndGet(ref.getMessageMemoryEstimate());
      pendingMetrics.incrementMetrics(ref);
      refAdded(ref);

      int priority = getPriority(ref);

      messageReferences.addHead(ref, priority);

      ref.setInDelivery(false);
   }

   //......
}

• QueueImpl的deliverScheduledMessages方法會執行internalAddHead，它會將scheduledMessages按priority添加到messageReferences的隊頭

小結

• CoreMessage的getScheduledDeliveryTime會讀取Message.HDR_SCHEDULED_DELIVERY_TIME屬性的值；setScheduledDeliveryTime方法則會更新Message.HDR_SCHEDULED_DELIVERY_TIME屬性的值
• PostOfficeImpl的processRoute方法會將message.getScheduledDeliveryTime()更新到reference.setScheduledDeliveryTime(deliveryTime)；對於isDurable爲true的message還會執行storageManager.storeMessage；對於tx不爲null的會建立AddOperation添加到tx中，對於tx爲null的則經過storageManager.afterCompleteOperations註冊IOCallback，其done方法執行context.processReferences(refs, direct)；AddOperation的afterCommit方法對於非isAlreadyAcked的會執行ref.getQueue().addTail(ref, false)；RoutingContextImpl的processReferences方法執行ref.getQueue().addTail(ref, direct)
• PostOfficeJournalLoader的handleAddMessage會執行postOffice.reroute；PostOfficeImpl的reroute方法對於tx爲null執行queue.reload(reference)，對於tx不爲null的則建立AddOperation添加到tx；QueueImpl的reload會先執行scheduledDeliveryHandler.checkAndSchedule(ref, true)，若返回false則執行internalAddTail(ref)；internalAddTail則執行messageReferences.addTail
• QueueImpl的addTail方法會先執行scheduleIfPossible，若爲true則提早返回；若爲false的話則繼續往下，將MessageReference添加到intermediateMessageReferences；scheduleIfPossible方法則執行scheduledDeliveryHandler.checkAndSchedule(ref, true)，其返回true則爲true，不然爲false；QueueImpl的doInternalPoll方法會將intermediateMessageReferences裏頭的message轉移到messageReferences中
• ScheduledDeliveryHandlerImpl的checkAndSchedule方法會執行scheduleDelivery，該方法對於delay小於0的會立馬往建立ScheduledDeliveryRunnable並提交scheduledExecutor調度；不然會維護deliveryTime以及ScheduledDeliveryRunnable的映射到runnables中，並往scheduledExecutor提交runnable延時執行；ScheduledDeliveryRunnable主要是對到期的MessageReference執行queue.addHead(list, true)(其中list爲LinkedList<MessageReference>)；ScaleDownHandler的scaleDownMessages方法會執行queue.deliverScheduledMessages()；QueueImpl的deliverScheduledMessages方法會執行internalAddHead，它會將scheduledMessages按priority添加到messageReferences的隊頭

doc

• QueueImpl