Configuring Queue Options
The C++ Broker M4 or later supports the following additional Queue constraints.
- Applying Queue Sizing Constraints
- Changing the Queue ordering Behaviors
- Setting additional behaviors
- Other Clients
Applying Queue Sizing Constraints
This allows to specify how to size a queue and what to do when the sizing constraints have been reached. The queue size can be limited by the number messages (message depth) or byte depth on the queue.
Once the Queue meets/ exceeds these constraints the follow policies can be applied
- REJECT - Reject the published message
- FLOW_TO_DISK - Flow the messages to disk, to preserve memory
- RING - start overwriting messages in a ring based on sizing. If head meets tail, advance head
- RING_STRICT - start overwriting messages in a ring based on sizing. If head meets tail, AND the consumer has the tail message acquired it will reject
Examples:
Create a queue an auto delete queue that will support 100 000 bytes, and then REJECT
#include "qpid/client/QueueOptions.h" QueueOptions qo; qo.setSizePolicy(REJECT,100000,0); session.queueDeclare(arg::queue=queue, arg::autoDelete=true, arg::arguments=qo);
Create a queue that will support 1000 messages into a RING buffer
#include "qpid/client/QueueOptions.h" QueueOptions qo; qo.setSizePolicy(RING,0,1000); session.queueDeclare(arg::queue=queue, arg::arguments=qo);
Changing the Queue ordering Behaviors
The default ordering in a queue in Qpid is FIFO. However additional ordering semantics can be used namely LVQ (Last Value Queue). Last Value Queue is define as follows.
If I publish symbols RHT, IBM, JAVA, MSFT, and then publish RHT before the consumer is able to consume RHT, that message will be over written in the queue and the consumer will receive the last published value for RHT.
Example:
#include "qpid/client/QueueOptions.h" QueueOptions qo; qo.setOrdering(LVQ); session.queueDeclare(arg::queue=queue, arg::arguments=qo); ..... string key; qo.getLVQKey(key); .... for each message, set the into application headers before transfer message.getHeaders().setString(key,"RHT");
Setting additional behaviors - Optimistic Consume
Optimistic Consume allows the consumer to dequeue the message before the broker has acknowledged the producer. This is useful to reduce latency when doing duarble messaging, but has the following additional consideration.
In a failure it is possible that the consumer will receive duplicate messages. These can come from two sources.
- The producer might re-send a message the consumer receiver, because the broker had not acknowledged the producer yet
- The broker might not have persisted the consumers dequeue, resulting in a duplicate resend.
Example:
#include "qpid/client/QueueOptions.h" QueueOptions qo; qo.setOptimisticConsume(); session.queueDeclare(arg::queue=queue, arg::arguments=qo);
Setting additional behaviors - Persist Last Node
This option is used in conjunction with clustering. It allows for a queue configured with this option to persist transient messages if the cluster fails down to the last node. If additional nodes in the cluster are restored it will stop persisting transient messages.
Note
- if a cluster is started with only one active node, this mode will not be triggered. It is only triggered the first time the cluster fails down to 1 node.
- Messages that are in the Queue are flushed to the store optimistically. New messages received honor the Optimistic mode that has been configured on the queue.
- The queue MUST be configured durable
Example:
#include "qpid/client/QueueOptions.h" QueueOptions qo; qo.clearPersistLastNode(); session.queueDeclare(arg::queue=queue, arg::durable=true, arg::arguments=qo);
Other Clients
Note that these options can be set from any client. QueueOptions just correctly formats the arguments passed to the QueueDeclare() method.