Distributed Priority Queue

Description

An implementation of the Distributed Priority Queue ZK recipe.

Participating Classes

• QueueBuilder
• QueueConsumer
• QueueSerializer
• DistributedPriorityQueue

Usage

Creating a DistributedPriorityQueue

public static  QueueBuilder builder(CuratorFramework client,
QueueConsumer consumer,
QueueSerializer serializer,
java.lang.String queuePath)
Parameters:
client - the curator client
consumer - message consumer
serializer - serializer to use for items
queuePath - path to store queue

QueueBuilder    builder = QueueBuilder.builder(client, consumer, serializer, path);
... more builder method calls as needed ...
DistributedPriorityQueue<MessageType queue = builder.buildPriorityQueue(minItemsBeforeRefresh);

public DistributedPriorityQueue buildPriorityQueue(int minItemsBeforeRefresh)
Build a DistributedPriorityQueue from the current builder values.

When the priority queue detects an item addition/removal, it will stop processing its current list of items and refresh the list. minItemsBeforeRefresh modifies this. It determines the minimum number of items from the active list that will get processed before a refresh.

Due to a quirk in the way ZooKeeper notifies changes, the queue will get an item addition/remove notification after every item is processed. This can lead to poor performance. Set minItemsBeforeRefresh to the value your application can tolerate being out of sync.

For example: if the queue sees 10 items to process, it will end up making 10 calls to ZooKeeper to check status. You can control this by setting minItemsBeforeRefresh to 10 (or more) and the queue will only refresh with ZooKeeper after 10 items are processed

Parameters: minItemsBeforeRefresh - minimum items to process before refreshing the item list

General Usage

The queue must be started via the start() method. Call close() when you are done with the queue.

To add messages to the queue:

queue.put(aMessage, priority);

The consumer will get called as messages arrive.