Package org.acemq.amqp.core

Class ConsumerGroup

java.lang.Object
org.acemq.amqp.core.ConsumerGroup
All Implemented Interfaces:
AutoCloseable
public final class ConsumerGroup extends Object implements AutoCloseable
Several consumers on one queue, resized while the application runs.

Two numbers govern how fast a queue drains, and both of them are usually guessed once, put in a properties file, and never revisited — because changing either has meant a redeploy. Neither actually has to.

  • Prefetch — how many messages one consumer holds unacknowledged. Too low and the consumer idles between round trips; too high and one consumer hoards the queue while its neighbours starve, and a restart redelivers all of it.
  • Concurrency — how many consumers there are. This is the one that adds throughput when handlers are slow because they are waiting on something else. 

 ConsumerGroup group = mq.consumeGroup("orders.new", Order.class, handler)
         .concurrency(4)
         .prefetch(50)
         .start();

 group.scaleTo(8);      // Friday afternoon
 group.prefetch(100);
 group.scaleTo(2);      // Sunday night, draining the four that go

Scaling down drains rather than cancels. Removing a consumer that is mid-message would leave the broker to redeliver it somewhere else, which is only harmless if every handler is idempotent — and if that were reliably true, half this library would not need to exist. So a consumer being removed stops taking new work and is given time to finish what it holds.

  • Method Details

    • queue

      public String queue()
      Returns:
      the queue every member consumes

    • size

      public int size()
      Returns:
      how many consumers are running

    • prefetch

      public int prefetch()
      Returns:
      the prefetch each member is using

    • drainTimeout

      public ConsumerGroup drainTimeout(Duration timeout)
      Parameters:
      timeout - how long the consumers being removed have between them to finish their work
      Returns:
      this group

    • scaleTo

      public ConsumerGroup scaleTo(int size)
      Grows or shrinks the group.

      Growing is immediate. Shrinking drains the consumers being removed, so this call blocks for as long as they need, up to drainTimeout(Duration). That is deliberate: a scale-down that returned instantly and left work being abandoned in the background would be the more convenient lie.

      Parameters:
      size - how many consumers there should be, at least one
      Returns:
      this group

    • prefetch

      public ConsumerGroup prefetch(int prefetch)
      Changes the prefetch of every member, and of members added later.
      Parameters:
      prefetch - the new limit, at least 1
      Returns:
      this group

    • inFlight

      public long inFlight()
      Returns:
      how many messages the whole group is handling right now

    • acknowledged

      public long acknowledged()
      Returns:
      how many messages the group has acknowledged

    • rejected

      public long rejected()
      Returns:
      how many deliveries the group has rejected

    • retried

      public long retried()
      Returns:
      how many deliveries the group has retried

    • drain

      public boolean drain(Duration timeout)
      Stops every member, letting them finish what they hold.
      Parameters:
      timeout - how long to wait in total, shared by every member
      Returns:
      whether everything finished in time

    • close

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

    • toString

      public String toString()
      Overrides:
      toString in class Object