com.github.ddth.queue.impl

Class DisruptorQueue

java.lang.Object

com.github.ddth.queue.impl.DisruptorQueue

All Implemented Interfaces:

IQueue, Closeable, AutoCloseable

Direct Known Subclasses:

BaseUniversalDisruptorQueue

public class DisruptorQueue
extends Object
implements IQueue, Closeable, AutoCloseable

In-Memory implementation of IQueue using LMAX Disruptor library.

Implementation:

• A Disruptor's RingBuffer to be the queue storage.
• A ConcurrentMap as ephemeral storage.

Since:

0.4.0

Author:

Thanh Ba Nguyen

See Also:

https://lmax-exchange.github.io/disruptor/

Constructor Summary

Constructors

Constructor and Description
DisruptorQueue()
DisruptorQueue(int ringSize)

Method Summary

Modifier and Type	Method and Description
void	close()
void	destroy() Destroy method.
int	ephemeralSize() Gets ephemeral-storage's number of items.
void	finish(IQueueMessage msg) Called when finish processing the message to cleanup ephemeral storage.
boolean	getEphemeralDisabled() Is ephemeral storage disabled?
Collection<IQueueMessage>	getOrphanMessages(long thresholdTimestampMs) Gets all orphan messages (messages that were left in ephemeral storage for a long time).
protected int	getRingSize() Gets size of the ring buffer.
DisruptorQueue	init() Init method.
boolean	isEphemeralDisabled() Is ephemeral storage disabled?
boolean	moveFromEphemeralToQueueStorage(IQueueMessage _msg) Moves a message from ephemeral back to queue storage.
protected void	publish(IQueueMessage value, long seq) Publishes (commits) the ring's sequence.
protected boolean	putToRingBuffer(IQueueMessage msg) Puts a message to the ring buffer.
boolean	queue(IQueueMessage _msg) Queues a message.
int	queueSize() Gets queue's number of items.
boolean	requeue(IQueueMessage _msg) Re-queues a message.
boolean	requeueSilent(IQueueMessage msg) Silently re-queues a message.
DisruptorQueue	setEphemeralDisabled(boolean ephemeralDisabled) Disables/Enables ephemeral storage.
DisruptorQueue	setRingSize(int ringSize) Sets size of the ring buffer, must be power of 2.
IQueueMessage	take() Takes a message out of queue.
protected IQueueMessage	takeFromRingBuffer() Takes a message from the ring buffer.
protected void	updatePublishedSequence() Updates the published sequence number after a message has been successfully taken from queue.

Methods inherited from class java.lang.Object

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

Constructor Detail

DisruptorQueue

public DisruptorQueue()

DisruptorQueue

public DisruptorQueue(int ringSize)

Method Detail

getEphemeralDisabled

public boolean getEphemeralDisabled()

Is ephemeral storage disabled?

Returns:

isEphemeralDisabled

public boolean isEphemeralDisabled()

Is ephemeral storage disabled?

Returns:

setEphemeralDisabled

public DisruptorQueue setEphemeralDisabled(boolean ephemeralDisabled)

Disables/Enables ephemeral storage.

Parameters:

ephemeralDisabled - true to disable ephemeral storage, false otherwise.

Returns:

getRingSize

protected int getRingSize()

Gets size of the ring buffer.

Returns:

setRingSize

public DisruptorQueue setRingSize(int ringSize)

Sets size of the ring buffer, must be power of 2.

Parameters:

ringSize -

Returns:

init

public DisruptorQueue init()

Init method.

Returns:

destroy

public void destroy()

Destroy method.

close

public void close()

Specified by:

close in interface Closeable

Specified by:

close in interface AutoCloseable

publish

protected void publish(IQueueMessage value,
                       long seq)

Publishes (commits) the ring's sequence.

Parameters:

value -

seq -

putToRingBuffer

protected boolean putToRingBuffer(IQueueMessage msg)

Puts a message to the ring buffer.

Parameters:

msg -

Returns:

true if successful, false otherwise (the ring buffer is full

queue

public boolean queue(IQueueMessage _msg)

Queues a message.

Implementation flow:

• Put message to tail of queue storage.

Specified by:

queue in interface IQueue

Returns:

requeue

public boolean requeue(IQueueMessage _msg)

Re-queues a message.

Implementation flow:

• Put message to tail of queue storage; and increase message's re-queue count & update message's queue timestamp.
• Remove message from ephemeral storage.

Note: ephemeral storage implementation is optional, depends on implementation.

Specified by:

requeue in interface IQueue

Returns:

requeueSilent

public boolean requeueSilent(IQueueMessage msg)

Silently re-queues a message.

Implementation flow:

• Put message to tail of queue storage; do NOT increase message's re-queue count and do NOT update message's queue timestamp.
• Remove message from ephemeral storage.

Note: ephemeral storage implementation is optional, depends on implementation.

Specified by:

requeueSilent in interface IQueue

Returns:

finish

public void finish(IQueueMessage msg)

Called when finish processing the message to cleanup ephemeral storage.

Implementation flow:

• Remove message from ephemeral storage.

Note: ephemeral storage implementation is optional, depends on implementation.

Specified by:

finish in interface IQueue

updatePublishedSequence

protected void updatePublishedSequence()

Updates the published sequence number after a message has been successfully taken from queue.

takeFromRingBuffer

protected IQueueMessage takeFromRingBuffer()

Takes a message from the ring buffer.

Returns:

the available message or null if the ring buffer is empty

take

public IQueueMessage take()

Takes a message out of queue.

Implementation flow:

• Read message from head of queue storage.
• Write message to ephemeral storage.
• Remove message from queue storage.

Note: ephemeral storage implementation is optional, depends on implementation.

Specified by:

take in interface IQueue

Returns:

getOrphanMessages

public Collection<IQueueMessage> getOrphanMessages(long thresholdTimestampMs)

Gets all orphan messages (messages that were left in ephemeral storage for a long time).

Specified by:

getOrphanMessages in interface IQueue

Parameters:

thresholdTimestampMs -

Returns:

moveFromEphemeralToQueueStorage

public boolean moveFromEphemeralToQueueStorage(IQueueMessage _msg)

Moves a message from ephemeral back to queue storage. Useful when dealing with orphan messages.

Implementation flow:

• Read message from the ephemeral storage.
• Put the message back to queue.
• Remove the message from the ephemeral storage.

Specified by:

moveFromEphemeralToQueueStorage in interface IQueue

Returns:

true if a move has been made, false otherwise (e.g. the message didn't exist in ephemeral storage)

queueSize

public int queueSize()

Gets queue's number of items.

Specified by:

queueSize in interface IQueue

Returns:

ephemeralSize

public int ephemeralSize()

Gets ephemeral-storage's number of items.

Note: ephemeral storage implementation is optional, depends on implementation.

Specified by:

ephemeralSize in interface IQueue

Returns: