com.github.ddth.queue.impl

Class JdbcQueue

java.lang.Object

com.github.ddth.dao.BaseDao

com.github.ddth.dao.jdbc.BaseJdbcDao

com.github.ddth.queue.impl.JdbcQueue

All Implemented Interfaces:

IQueue

Direct Known Subclasses:

LessLockingUniversalMySQLQueue, LessLockingUniversalPgSQLQueue, UniversalJdbcQueue

public abstract class JdbcQueue
extends com.github.ddth.dao.jdbc.BaseJdbcDao
implements IQueue

Abstract JDBC implementation of IQueue.

Implementation:

• Queue storage & Ephemeral storage are 2 database tables, same structure!

Since:

0.1.0

Author:

Thanh Ba Nguyen

Field Summary

Fields

Modifier and Type	Field and Description
static int	DEFAULT_MAX_RETRIES

Fields inherited from class com.github.ddth.dao.BaseDao

CHARSET

Constructor Summary

Constructors

Constructor and Description
JdbcQueue()

Method Summary

Methods

Modifier and Type	Method and Description
protected void	_finishWithRetries(Connection conn, IQueueMessage msg, int numRetries, int maxRetries) Performs "finish" action, retry if deadlock.
protected Collection<IQueueMessage>	_getOrphanMessagesWithRetries(long thresholdTimestampMs, Connection conn, int numRetries, int maxRetries) Gets all orphan messages (messages that were left in ephemeral storage for a long time), retry if deadlock.
protected boolean	_moveFromEphemeralToQueueStorageWithRetries(IQueueMessage msg, Connection conn, int numRetries, int maxRetries) Moves a message from ephemeral back to queue storage, retry if deadlock.
protected boolean	_queueWithRetries(Connection conn, IQueueMessage msg, int numRetries, int maxRetries) Queues a message, retry if deadlock.
protected boolean	_requeueSilentWithRetries(Connection conn, IQueueMessage msg, int numRetries, int maxRetries) Re-queues a message silently, retry if deadlock.
protected boolean	_requeueWithRetries(Connection conn, IQueueMessage msg, int numRetries, int maxRetries) Re-queues a message, retry if deadlock.
protected IQueueMessage	_takeWithRetries(Connection conn, int numRetries, int maxRetries) Takes a message from queue, retry if deadlock.
void	destroy()
int	ephemeralSize() Gets ephemeral-storage's number of items.
void	finish(IQueueMessage msg) Called when finish processing the message to cleanup ephemeral storage.
int	getMaxRetries()
protected abstract Collection<IQueueMessage>	getOrphanFromEphemeralStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, long thresholdTimestampMs) Gets all orphan messages (messages that were left in ephemeral storage for a long time).
Collection<IQueueMessage>	getOrphanMessages(long thresholdTimestampMs) Gets all orphan messages (messages that were left in ephemeral storage for a long time).
String	getTableName()
String	getTableNameEphemeral()
int	getTransactionIsolationLevel()
JdbcQueue	init()
boolean	moveFromEphemeralToQueueStorage(IQueueMessage msg) Moves a message from ephemeral back to queue storage.
protected abstract boolean	putToEphemeralStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage msg) Puts a message to the ephemeral storage.
protected abstract boolean	putToQueueStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage msg) Puts a message to tail of the queue storage.
boolean	queue(IQueueMessage msg) Queues a message.
int	queueSize() Gets queue's number of items.
protected abstract IQueueMessage	readFromEphemeralStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage msg) Reads a message from the ephemeral storage.
protected abstract IQueueMessage	readFromQueueStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate) Reads a message from head of queue storage.
protected abstract boolean	removeFromEphemeralStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage msg) Removes a message from the ephemeral storage.
protected abstract boolean	removeFromQueueStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage msg) Removes a message from the queue storage.
boolean	requeue(IQueueMessage msg) Re-queues a message.
boolean	requeueSilent(IQueueMessage msg) Silently re-queues a message.
JdbcQueue	setMaxRetries(int maxRetries)
JdbcQueue	setTableName(String tableName)
JdbcQueue	setTableNameEphemeral(String tableNameEphemeral)
JdbcQueue	setTransactionIsolationLevel(int transactionIsolationLevel)
IQueueMessage	take() Takes a message out of queue.

Methods inherited from class com.github.ddth.dao.jdbc.BaseJdbcDao

commitTransaction, connection, connection, execute, execute, execute, executeSelect, executeSelect, executeSelect, executeSelect, executeSelect, executeSelect, getDataSource, jdbcTemplate, jdbcTemplate, returnConnection, rollbackTransaction, setDataSource, startTransaction

Methods inherited from class com.github.ddth.dao.BaseDao

addProfiling, clearProfiling, getCache, getCacheFactory, getFromCache, getFromCache, getProfiling, isCacheEnabled, isCacheItemsExpireAfterWrite, putToCache, putToCache, putToCache, removeFromCache, setCacheFactory, setCacheItemsExpireAfterWrite, startProfiling

Methods inherited from class java.lang.Object

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

Field Detail

DEFAULT_MAX_RETRIES

public static int DEFAULT_MAX_RETRIES

Constructor Detail

JdbcQueue

public JdbcQueue()

Method Detail

setTableName

public JdbcQueue setTableName(String tableName)

getTableName

public String getTableName()

setTableNameEphemeral

public JdbcQueue setTableNameEphemeral(String tableNameEphemeral)

getTableNameEphemeral

public String getTableNameEphemeral()

setTransactionIsolationLevel

public JdbcQueue setTransactionIsolationLevel(int transactionIsolationLevel)

getTransactionIsolationLevel

public int getTransactionIsolationLevel()

setMaxRetries

public JdbcQueue setMaxRetries(int maxRetries)

getMaxRetries

public int getMaxRetries()

init

public JdbcQueue init()

Overrides:

init in class com.github.ddth.dao.jdbc.BaseJdbcDao

destroy

public void destroy()

Overrides:

destroy in class com.github.ddth.dao.jdbc.BaseJdbcDao

readFromQueueStorage

protected abstract IQueueMessage readFromQueueStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate)

Reads a message from head of queue storage.

Parameters:

jdbcTemplate -

Returns:

readFromEphemeralStorage

protected abstract IQueueMessage readFromEphemeralStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate,
                                     IQueueMessage msg)

Reads a message from the ephemeral storage.

Parameters:

jdbcTemplate -

msg -

Returns:

Since:

0.2.1

getOrphanFromEphemeralStorage

protected abstract Collection<IQueueMessage> getOrphanFromEphemeralStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate,
                                                      long thresholdTimestampMs)

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

Parameters:

jdbcTemplate -

thresholdTimestampMs - get all orphan messages that were queued before this timestamp

Returns:

Since:

0.2.0

putToQueueStorage

protected abstract boolean putToQueueStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate,
                        IQueueMessage msg)

Puts a message to tail of the queue storage.

Parameters:

jdbcTemplate -

msg -

Returns:

putToEphemeralStorage

protected abstract boolean putToEphemeralStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate,
                            IQueueMessage msg)

Puts a message to the ephemeral storage.

Parameters:

jdbcTemplate -

msg -

Returns:

removeFromQueueStorage

protected abstract boolean removeFromQueueStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate,
                             IQueueMessage msg)

Removes a message from the queue storage.

Parameters:

jdbcTemplate -

msg -

Returns:

removeFromEphemeralStorage

protected abstract boolean removeFromEphemeralStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate,
                                 IQueueMessage msg)

Removes a message from the ephemeral storage.

Parameters:

jdbcTemplate -

msg -

Returns:

_queueWithRetries

protected boolean _queueWithRetries(Connection conn,
                        IQueueMessage msg,
                        int numRetries,
                        int maxRetries)
                             throws SQLException

Queues a message, retry if deadlock.

Note: http://dev.mysql.com/doc/refman/5.0/en/innodb-deadlocks.html

InnoDB uses automatic row-level locking. You can get deadlocks even in the case of transactions that just insert or delete a single row. That is because these operations are not really "atomic"; they automatically set locks on the (possibly several) index records of the row inserted or deleted.

Parameters:

conn -

msg -

numRetries -

maxRetries -

Returns:

Throws:

SQLException

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:

_requeueWithRetries

protected boolean _requeueWithRetries(Connection conn,
                          IQueueMessage msg,
                          int numRetries,
                          int maxRetries)
                               throws SQLException

Re-queues a message, retry if deadlock.

Note: http://dev.mysql.com/doc/refman/5.0/en/innodb-deadlocks.html

InnoDB uses automatic row-level locking. You can get deadlocks even in the case of transactions that just insert or delete a single row. That is because these operations are not really "atomic"; they automatically set locks on the (possibly several) index records of the row inserted or deleted.

Parameters:

conn -

msg -

numRetries -

maxRetries -

Returns:

Throws:

SQLException

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:

_requeueSilentWithRetries

protected boolean _requeueSilentWithRetries(Connection conn,
                                IQueueMessage msg,
                                int numRetries,
                                int maxRetries)
                                     throws SQLException

Re-queues a message silently, retry if deadlock.

Note: http://dev.mysql.com/doc/refman/5.0/en/innodb-deadlocks.html

InnoDB uses automatic row-level locking. You can get deadlocks even in the case of transactions that just insert or delete a single row. That is because these operations are not really "atomic"; they automatically set locks on the (possibly several) index records of the row inserted or deleted.

Parameters:

conn -

msg -

numRetries -

maxRetries -

Returns:

Throws:

SQLException

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:

_finishWithRetries

protected void _finishWithRetries(Connection conn,
                      IQueueMessage msg,
                      int numRetries,
                      int maxRetries)
                           throws SQLException

Performs "finish" action, retry if deadlock.

Note: http://dev.mysql.com/doc/refman/5.0/en/innodb-deadlocks.html

InnoDB uses automatic row-level locking. You can get deadlocks even in the case of transactions that just insert or delete a single row. That is because these operations are not really "atomic"; they automatically set locks on the (possibly several) index records of the row inserted or deleted.

Parameters:

conn -

msg -

numRetries -

maxRetries -

Throws:

SQLException

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

_takeWithRetries

protected IQueueMessage _takeWithRetries(Connection conn,
                             int numRetries,
                             int maxRetries)
                                  throws SQLException

Takes a message from queue, retry if deadlock.

Note: http://dev.mysql.com/doc/refman/5.0/en/innodb-deadlocks.html

InnoDB uses automatic row-level locking. You can get deadlocks even in the case of transactions that just insert or delete a single row. That is because these operations are not really "atomic"; they automatically set locks on the (possibly several) index records of the row inserted or deleted.

Parameters:

conn -

numRetries -

maxRetries -

Returns:

Throws:

SQLException

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:

_getOrphanMessagesWithRetries

protected Collection<IQueueMessage> _getOrphanMessagesWithRetries(long thresholdTimestampMs,
                                                      Connection conn,
                                                      int numRetries,
                                                      int maxRetries)
                                                           throws SQLException

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

Note: http://dev.mysql.com/doc/refman/5.0/en/innodb-deadlocks.html

InnoDB uses automatic row-level locking. You can get deadlocks even in the case of transactions that just insert or delete a single row. That is because these operations are not really "atomic"; they automatically set locks on the (possibly several) index records of the row inserted or deleted.

Parameters:

thresholdTimestampMs -

conn -

numRetries -

maxRetries -

Returns:

Throws:

SQLException

Since:

0.2.0

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 - get all orphan messages that were queued before this timestamp

Returns:

_moveFromEphemeralToQueueStorageWithRetries

protected boolean _moveFromEphemeralToQueueStorageWithRetries(IQueueMessage msg,
                                                  Connection conn,
                                                  int numRetries,
                                                  int maxRetries)
                                                       throws SQLException

Moves a message from ephemeral back to queue storage, retry if deadlock.

Note: http://dev.mysql.com/doc/refman/5.0/en/innodb-deadlocks.html

InnoDB uses automatic row-level locking. You can get deadlocks even in the case of transactions that just insert or delete a single row. That is because these operations are not really "atomic"; they automatically set locks on the (possibly several) index records of the row inserted or deleted.

Parameters:

msg -

conn -

numRetries -

maxRetries -

Returns:

Throws:

SQLException

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: