Package org.broadleafcommerce.core.util.queue

Class ZookeeperDistributedQueue<T extends Serializable>

java.lang.Object

org.broadleafcommerce.core.util.queue.ZookeeperDistributedQueue<T>

All Implemented Interfaces:

Iterable<T>, Collection<T>, BlockingQueue<T>, Queue<T>, DistributedBlockingQueue<T>

public class ZookeeperDistributedQueue<T extends Serializable>
extends Object
implements DistributedBlockingQueue<T>

Represents a Queue that is distributed (used by multiple JVMs or nodes) and managed by Zookeeper. This queue uses distributed locks, also backed by Zookeeper.

Please note that while this works quite well in certain circumstances, it is not recommended for high volume or high capacity queues, nor for large queue messages. It's a relatively slow queue. Zookeeper allows you to create queues that can be used in a distributed way, but large queues can cause performance problems in Zookeeper, and Zookeeper has a 1MB transport limit, so messages have to be smaller than that. Incidentally, initial performance tests showed queue operations (put / take) taking approximately 25-30 milliseconds, or about 30-40 queue operations per second with a small payload (about 15 bytes).

This Queue works quite well for smaller, lower capacity / throughput queues where you need to read/write in a distributed way. Try to limit the size of this queue to around 500 elements or fewer. Otherwise, consider a different queue implementation.

Author:

Kelly Tisdell

Nested Class Summary

Nested classes/interfaces inherited from interface org.broadleafcommerce.core.util.queue.DistributedBlockingQueue

DistributedBlockingQueue.DistributedQueueException

Field Summary

Fields

Modifier and Type	Field	Description
static final String	DEFAULT_BASE_FOLDER	This is the base folder that all queues will be written to in Solr.
static final int	DEFAULT_MAX_QUEUE_SIZE
static final String	QUEUE_CONFIGS_FOLDER
static final String	QUEUE_ENTRY_FOLDER
static final String	QUEUE_LOCKS_FOLDER
protected final Object	QUEUE_MONITOR

Constructor Summary

Constructors

Constructor	Description
ZookeeperDistributedQueue(String queuePath, org.apache.zookeeper.ZooKeeper zk)	Constructs a folder structure in Zookeeper for managing a queue and queue state..
ZookeeperDistributedQueue(String queuePath, org.apache.zookeeper.ZooKeeper zk, int maxQueueSize)	Constructs a folder structure in Zookeeper for managing a queue and queue state..
ZookeeperDistributedQueue(String queuePath, org.apache.zookeeper.ZooKeeper zk, int maxQueueSize, boolean useDefaultBasePath, List<org.apache.zookeeper.data.ACL> acls)	Constructs a folder structure in Zookeeper for managing a queue and queue state..

Method Summary

Modifier and Type	Method	Description
boolean	add(T e)
boolean	addAll(Collection<? extends T> c)
void	clear()
boolean	contains(Object o)
boolean	containsAll(Collection<?> c)
protected Object	deserialize(byte[] bytes)	Mechanism to convert a byte array to an object.
protected void	determineMaxCapacity()
int	drainTo(Collection<? super T> c)
int	drainTo(Collection<? super T> c, int maxElements)
T	element()
protected <R> R	executeOperation(GenericOperation<R> operation)	Allows us to execute retry-able operations.
protected int	geMaxCapacity()
protected List<org.apache.zookeeper.data.ACL>	getAcls()
protected DistributedLock	getConfigLock()
protected String	getConfigsFolder()
protected String	getLocksFolder()
protected DistributedLock	getQueueAccessLock()
protected String	getQueueEntryFolder()
protected String	getQueueEntryName()
String	getQueueFolderPath()
protected int	getRequestedMaxQueueSize()
protected org.apache.zookeeper.ZooKeeper	getZookeeperClient()
protected DistributedLock	initializeConfigLock()
protected DistributedLock	initializeQueueAccessLock()
protected void	intializeQueueFolders()	Creates the appropriate folder(s) in Zookeeper if they don't already exist.
boolean	isEmpty()
Iterator<T>	iterator()
boolean	offer(T e)
boolean	offer(T e, long timeout, TimeUnit unit)
T	peek()
T	poll()
T	poll(long timeout, TimeUnit unit)
void	put(T e)
protected Map<String,T>	readQueueInternal(int qty, boolean remove, long timeout)
int	remainingCapacity()
T	remove()
boolean	remove(Object o)
boolean	removeAll(Collection<?> c)
boolean	retainAll(Collection<?> c)
protected void	seMaxCapacity(int size)
protected byte[]	serialize(Serializable obj)	Mechanism to convert an object to a byte array.
int	size()
T	take()
Object[]	toArray()
<T> T[]	toArray(T[] a)
protected int	writeToQueue(List<? extends T> entries, long timeout)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface java.util.Collection

equals, hashCode, parallelStream, removeIf, spliterator, stream, toArray

Methods inherited from interface java.lang.Iterable

forEach

Field Details

DEFAULT_BASE_FOLDER

public static final String DEFAULT_BASE_FOLDER

This is the base folder that all queues will be written to in Solr. The constructors require a lock path, which will be appended to this path.

See Also:

• Constant Field Values

QUEUE_ENTRY_FOLDER

public static final String QUEUE_ENTRY_FOLDER

See Also:

• Constant Field Values

QUEUE_LOCKS_FOLDER

public static final String QUEUE_LOCKS_FOLDER

See Also:

• Constant Field Values

QUEUE_CONFIGS_FOLDER

public static final String QUEUE_CONFIGS_FOLDER

See Also:

• Constant Field Values

DEFAULT_MAX_QUEUE_SIZE

public static final int DEFAULT_MAX_QUEUE_SIZE

See Also:

• Constant Field Values

QUEUE_MONITOR

protected final Object QUEUE_MONITOR

Constructor Details

ZookeeperDistributedQueue

public ZookeeperDistributedQueue(String queuePath,
 org.apache.zookeeper.ZooKeeper zk)

Constructs a folder structure in Zookeeper for managing a queue and queue state.. The argument, queuePath, should start with a forward slash ('/') and should not end with a slash. This argument should not contain whitespaces or other special characters.

The default max queue size will be 500.

Parameters:

queuePath -

zk -

ZookeeperDistributedQueue

public ZookeeperDistributedQueue(String queuePath,
 org.apache.zookeeper.ZooKeeper zk,
 int maxQueueSize)

Constructs a folder structure in Zookeeper for managing a queue and queue state.. The argument, queuePath, should start with a forward slash ('/') and should not end with a slash. This argument should not contain whitespaces or other special characters.

The default max queue size will be 500.

Parameters:

queuePath -

zk -

maxQueueSize -

ZookeeperDistributedQueue

public ZookeeperDistributedQueue(String queuePath,
 org.apache.zookeeper.ZooKeeper zk,
 int maxQueueSize,
 boolean useDefaultBasePath,
 List<org.apache.zookeeper.data.ACL> acls)

Constructs a folder structure in Zookeeper for managing a queue and queue state.. The argument, queuePath, should start with a forward slash ('/') and should not end with a slash. This argument should be alpha-numeric, not contain whitespaces or other special characters, and can contain forward slashes ('/') to delineate folders. If useDefaultBasePath is true, then /broadleaf/app/distributed-queues will be prepended to the queuePath. Otherwise, the queuePath will be used as it is provided.

The argument, maxQueueSize, will be a hint. If another thread creates the queue structure in Zookeeper, then it will persist the maxQueueSize.

Parameters:

queuePath -

zk -

maxQueueSize -

useDefaultBasePath -

acls -

Method Details

remove

public T remove()

Specified by:

remove in interface Queue<T extends Serializable>

poll

public T poll()

Specified by:

poll in interface Queue<T extends Serializable>

element

public T element()

Specified by:

element in interface Queue<T extends Serializable>

peek

public T peek()

Specified by:

peek in interface Queue<T extends Serializable>

size

public int size()

Specified by:

size in interface Collection<T extends Serializable>

isEmpty

public boolean isEmpty()

Specified by:

isEmpty in interface Collection<T extends Serializable>

iterator

public Iterator<T> iterator()

Specified by:

iterator in interface Collection<T extends Serializable>

Specified by:

iterator in interface Iterable<T extends Serializable>

toArray

public Object[] toArray()

Specified by:

toArray in interface Collection<T extends Serializable>

toArray

public <T> T[] toArray(T[] a)

Specified by:

toArray in interface Collection<T extends Serializable>

containsAll

public boolean containsAll(Collection<?> c)

Specified by:

containsAll in interface Collection<T extends Serializable>

addAll

public boolean addAll(Collection<? extends T> c)

Specified by:

addAll in interface Collection<T extends Serializable>

removeAll

public boolean removeAll(Collection<?> c)

Specified by:

removeAll in interface Collection<T extends Serializable>

retainAll

public boolean retainAll(Collection<?> c)

Specified by:

retainAll in interface Collection<T extends Serializable>

clear

public void clear()

Specified by:

clear in interface Collection<T extends Serializable>

add

public boolean add(T e)

Specified by:

add in interface BlockingQueue<T extends Serializable>

Specified by:

add in interface Collection<T extends Serializable>

Specified by:

add in interface Queue<T extends Serializable>

offer

public boolean offer(T e)

Specified by:

offer in interface BlockingQueue<T extends Serializable>

Specified by:

offer in interface Queue<T extends Serializable>

put

public void put(T e)
         throws InterruptedException

Specified by:

put in interface BlockingQueue<T extends Serializable>

Throws:

InterruptedException

offer

public boolean offer(T e,
 long timeout,
 TimeUnit unit)
              throws InterruptedException

Specified by:

offer in interface BlockingQueue<T extends Serializable>

Throws:

InterruptedException

take

public T take()
       throws InterruptedException

Specified by:

take in interface BlockingQueue<T extends Serializable>

Throws:

InterruptedException

poll

public T poll(long timeout,
 TimeUnit unit)
       throws InterruptedException

Specified by:

poll in interface BlockingQueue<T extends Serializable>

Throws:

InterruptedException

remainingCapacity

public int remainingCapacity()

Specified by:

remainingCapacity in interface BlockingQueue<T extends Serializable>

remove

public boolean remove(Object o)

Specified by:

remove in interface BlockingQueue<T extends Serializable>

Specified by:

remove in interface Collection<T extends Serializable>

contains

public boolean contains(Object o)

Specified by:

contains in interface BlockingQueue<T extends Serializable>

Specified by:

contains in interface Collection<T extends Serializable>

drainTo

public int drainTo(Collection<? super T> c)

Specified by:

drainTo in interface BlockingQueue<T extends Serializable>

drainTo

public int drainTo(Collection<? super T> c,
 int maxElements)

Specified by:

drainTo in interface BlockingQueue<T extends Serializable>

writeToQueue

protected int writeToQueue(List<? extends T> entries,
 long timeout)
                    throws InterruptedException

Throws:

InterruptedException

readQueueInternal

protected Map<String,T> readQueueInternal(int qty,
 boolean remove,
 long timeout)
                                   throws InterruptedException

Throws:

InterruptedException

intializeQueueFolders

protected void intializeQueueFolders()

Creates the appropriate folder(s) in Zookeeper if they don't already exist.

determineMaxCapacity

protected void determineMaxCapacity()

deserialize

protected Object deserialize(byte[] bytes)

Mechanism to convert a byte array to an object. Default implementation uses ObjectInputStream.

Parameters:

bytes -

Returns:

serialize

protected byte[] serialize(Serializable obj)

Mechanism to convert an object to a byte array. Default implementation uses ObjectOutputStream.

Parameters:

obj -

Returns:

initializeQueueAccessLock

protected DistributedLock initializeQueueAccessLock()

initializeConfigLock

protected DistributedLock initializeConfigLock()

getRequestedMaxQueueSize

protected int getRequestedMaxQueueSize()

getQueueFolderPath

public String getQueueFolderPath()

getLocksFolder

protected String getLocksFolder()

getConfigsFolder

protected String getConfigsFolder()

getQueueEntryFolder

protected String getQueueEntryFolder()

geMaxCapacity

protected int geMaxCapacity()

seMaxCapacity

protected void seMaxCapacity(int size)

getQueueAccessLock

protected DistributedLock getQueueAccessLock()

getConfigLock

protected DistributedLock getConfigLock()

getQueueEntryName

protected String getQueueEntryName()

getZookeeperClient

protected org.apache.zookeeper.ZooKeeper getZookeeperClient()

getAcls

protected List<org.apache.zookeeper.data.ACL> getAcls()

executeOperation

protected <R> R executeOperation(GenericOperation<R> operation)
                          throws InterruptedException

Allows us to execute retry-able operations.

Parameters:

operation -

Returns:

Throws:

InterruptedException