Package org.acemq.amqp.patterns

Class Scheduler

java.lang.Object
org.acemq.amqp.patterns.Scheduler
All Implemented Interfaces:
AutoCloseable
public final class Scheduler extends Object implements AutoCloseable
Delivering a message later. 

 try (Scheduler scheduler = Scheduler.on(mq)) {
     scheduler.in(Duration.ofHours(4), "billing", "invoice.due", invoice);
     scheduler.at(renewalDate, "policies", "policy.renew", policy);
 }

Why not a per-message time to live

The obvious implementation is to set expiration on the message, drop it in a queue nobody consumes, and let it dead-letter to its destination. It is what most articles suggest and it is wrong for anything but a single fixed delay, because a classic queue expires messages only at its head.

Put a four-hour message in, then a one-minute message behind it, and the one-minute message is delivered in four hours. Nothing reports this: the queue looks healthy, the message is not lost, it is simply late by a factor nobody predicted. It is the single most common way a home-made scheduler fails, and it fails in production under mixed load rather than in testing under uniform load.

What this does instead

A small ladder of queues, each with a uniform time to live, and a message hops through them until it is due: 

 acemq.schedule.1s  acemq.schedule.10s  acemq.schedule.1m  acemq.schedule.10m  acemq.schedule.1h

Every message in a given rung has the same delay, so head-of-line expiry is not a problem — the head is always the message due soonest. Each expiry returns the message to this scheduler, which either delivers it or puts it in the largest rung that does not overshoot. A four-hour delay is four one-hour hops; a ninety-second delay is one minute, then three tens.

The cost is honest and worth stating: a long delay is several broker round trips rather than one, and delivery is accurate to about the smallest rung rather than to the second. A scheduler that must fire at 09:00:00.000 exactly is a scheduler, not a message broker.

The alternative is RabbitMQ's delayed-message-exchange plugin, which does this properly and is a plugin — so it is not available everywhere, and a library that silently required it would be a library that works on your laptop.

  • Method Details

    • on

      public static Scheduler on(AceMq mq)
      Parameters:
      mq - an open connection
      Returns:
      a scheduler, with its queues declared

    • in

      public void in(Duration delay, String exchange, String routingKey, Object payload)
      Delivers a message after a delay.
      Parameters:
      delay - how long to wait; zero or negative delivers immediately
      exchange - where it should eventually go
      routingKey - the routing key it should eventually carry
      payload - the message

    • at

      public void at(Instant when, String exchange, String routingKey, Object payload)
      Delivers a message at a moment.
      Parameters:
      when - the moment; anything in the past delivers immediately
      exchange - where it should eventually go
      routingKey - the routing key it should eventually carry
      payload - the message

    • scheduled

      public long scheduled()
      Returns:
      messages handed to this scheduler

    • delivered

      public long delivered()
      Returns:
      messages that reached their destination

    • hops

      public long hops()
      Returns:
      how many times a message moved between rungs. Divided by delivered() this is the average number of hops, which is the number to look at if the scheduler is busier than expected: long delays cost hops

    • close

      public void close() throws Exception
      Specified by:
      close in interface AutoCloseable
      Throws:
      Exception

    • toString

      public String toString()
      Overrides:
      toString in class Object