Package org.acemq.amqp.core

Class OrderedQueue<T>

java.lang.Object
org.acemq.amqp.core.OrderedQueue<T>
Type Parameters:
T - payload type
All Implemented Interfaces:
AutoCloseable
public final class OrderedQueue<T> extends Object implements AutoCloseable
Messages that share a key are handled in the order they were sent. 

 try (OrderedQueue<Order> orders = mq.ordered("orders", Order.class)
         .partitions(8)
         .keyedBy(Order::customerId)
         .declare()) {

     orders.send(order);
     orders.consume(message -> ledger.post(message.payload()));
 }

What this actually costs

Ordering and parallelism are opposed, and no amount of API hides that. What this does is make the standard trade cheap: the key decides a partition, each partition is a queue, and each queue has exactly one consumer. Eight partitions gives eight-way parallelism and ordering within any single key — which is almost always the ordering anyone actually wanted. Two orders for one customer are sequenced; two orders for different customers are not, and never needed to be.

Throughput is therefore bounded by the busiest key, not by the number of partitions. One very hot customer is one queue, one consumer, and no way to go faster without giving up the guarantee.

Retries reorder, so the retry ladder is not available here

This is the part usually left unsaid. If message five fails and is republished to come back in thirty seconds, message six is handled first and the sequence is broken. Every ordered consumer has to choose between blocking the partition and breaking the order, so OrderedQueue.OnFailure makes the choice explicit and offers nothing that quietly does neither.

RETRY_IN_PLACE sleeps in the handler, which is a bug in an unordered consumer and is the right answer here: the partition is meant to stall, because the alternative is delivering message six before message five.

  • Method Details

    • queueFor

      public String queueFor(int partition)
      Returns:
      the queue backing a partition

    • name

      public String name()
      Returns:
      the logical name, which is also the exchange

    • partitions

      public int partitions()
      Returns:
      how many partitions there are

    • send

      public int send(T payload)
      Parameters:
      payload - the message
      Returns:
      which partition it went to, which is worth having in a log when a sequence looks wrong

    • send

      public int send(T payload, @Nullable Envelope envelope)
      Parameters:
      payload - the message
      envelope - metadata to publish with, or null
      Returns:
      the partition it went to

    • consume

      public OrderedQueue<T> consume(MessageHandler<T> handler)
      Starts one consumer per partition.

      One, deliberately. A second consumer on a partition would take the next message while the first was still working, and the ordering would be gone.

      Parameters:
      handler - called for each message, in order within any one key
      Returns:
      this queue

    • haltedPartitions

      public Set<Integer> haltedPartitions()
      Returns:
      partitions stopped by a handler failure

    • handled

      public long handled()
      Returns:
      how many messages have been handled successfully

    • failed

      public long failed()
      Returns:
      how many handler attempts have failed

    • skipped

      public long skipped()
      Returns:
      how many messages were given up on; only ever non-zero with OrderedQueue.OnFailure.SKIP

    • queues

      public List<String> queues()
      Returns:
      the queues behind the partitions, in order

    • close

      public void close()
      Specified by:
      close in interface AutoCloseable

    • toString

      public String toString()
      Overrides:
      toString in class Object