com.github.ddth.queue.impl.universal

Class LessLockingUniversalMySQLQueue

java.lang.Object

com.github.ddth.dao.BaseDao

com.github.ddth.dao.jdbc.BaseJdbcDao

com.github.ddth.queue.impl.JdbcQueue

com.github.ddth.queue.impl.universal.LessLockingUniversalMySQLQueue

All Implemented Interfaces:

IQueue, Closeable, AutoCloseable

public class LessLockingUniversalMySQLQueue
extends JdbcQueue

Same as UniversalJdbcQueue, but using a less-locking algorithm - specific for MySQL, and requires only one single db table for both queue and ephemeral storages.

Queue db table schema:

• queue_id: bigint, auto increment, see IQueueMessage.qId(), COL_QUEUE_ID
• ephemeral_id: bigint, see COL_EPHEMERAL_ID
• msg_org_timestamp: datetime, see IQueueMessage.qOriginalTimestamp(), COL_ORG_TIMESTAMP
• msg_timestamp: datetime, see IQueueMessage.qTimestamp(), COL_TIMESTAMP
• msg_num_requeues: int, see IQueueMessage.qNumRequeues(), COL_NUM_REQUEUES
• msg_content: blob, message's content, see COL_CONTENT

Since:

0.2.3

Author:

Thanh Ba Nguyen

Field Summary

Fields

Modifier and Type	Field and Description
static String	COL_CONTENT Table's column name to store message's content
static String	COL_EPHEMERAL_ID Table's column name to store ephemeral id
static String	COL_NUM_REQUEUES Table's column name to store message's number of requeues
static String	COL_ORG_TIMESTAMP Table's column name to store message's original timestamp
static String	COL_QUEUE_ID Table's column name to store queue-id
static String	COL_TIMESTAMP Table's column name to store message's timestamp

Fields inherited from class com.github.ddth.queue.impl.JdbcQueue

DEFAULT_MAX_RETRIES

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

CHARSET

Constructor Summary

Constructors

Constructor and Description
LessLockingUniversalMySQLQueue()

Method Summary

Modifier and Type	Method and Description
protected void	_finishWithRetries(Connection conn, IQueueMessage _msg, int numRetries, int maxRetries) Performs "finish" action, 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 UniversalQueueMessage	_takeWithRetries(Connection conn, int numRetries, int maxRetries) Takes a message from queue, retry if deadlock.
boolean	getFifo() If true, queue message with lower id is ensured to be taken first.
protected 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).
String	getTableNameEphemeral()
LessLockingUniversalMySQLQueue	init()
boolean	isFifo() If true, queue message with lower id is ensured to be taken first.
LessLockingUniversalMySQLQueue	markFifo(boolean fifo) When set to true, queue message with lower id is ensured to be taken first.
protected boolean	putToEphemeralStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage _msg) Puts a message to the ephemeral storage.
protected boolean	putToQueueStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage _msg) Puts a message to tail of the queue storage.
protected UniversalQueueMessage	readFromEphemeralStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage msg) Reads a message from the ephemeral storage.
protected UniversalQueueMessage	readFromQueueStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate) Reads a message from head of queue storage.
protected boolean	removeFromEphemeralStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage _msg) Removes a message from the ephemeral storage.
protected boolean	removeFromQueueStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate, IQueueMessage msg) Removes a message from the queue storage.
LessLockingUniversalMySQLQueue	setFifo(boolean fifo) When set to true, queue message with lower id is ensured to be taken first.
UniversalQueueMessage	take() Takes a message out of queue.

Methods inherited from class com.github.ddth.queue.impl.JdbcQueue

_getOrphanMessagesWithRetries, close, destroy, ephemeralSize, finish, getMaxRetries, getOrphanMessages, getTableName, getTransactionIsolationLevel, moveFromEphemeralToQueueStorage, queue, queueSize, requeue, requeueSilent, setMaxRetries, setTableName, setTableNameEphemeral, setTransactionIsolationLevel

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

COL_QUEUE_ID

public static final String COL_QUEUE_ID

Table's column name to store queue-id

See Also:

Constant Field Values

COL_EPHEMERAL_ID

public static final String COL_EPHEMERAL_ID

Table's column name to store ephemeral id

See Also:

Constant Field Values

COL_ORG_TIMESTAMP

public static final String COL_ORG_TIMESTAMP

Table's column name to store message's original timestamp

See Also:

Constant Field Values

COL_TIMESTAMP

public static final String COL_TIMESTAMP

Table's column name to store message's timestamp

See Also:

Constant Field Values

COL_NUM_REQUEUES

public static final String COL_NUM_REQUEUES

Table's column name to store message's number of requeues

See Also:

Constant Field Values

COL_CONTENT

public static final String COL_CONTENT

Table's column name to store message's content

See Also:

Constant Field Values

Constructor Detail

LessLockingUniversalMySQLQueue

public LessLockingUniversalMySQLQueue()

Method Detail

setFifo

public LessLockingUniversalMySQLQueue setFifo(boolean fifo)

When set to true, queue message with lower id is ensured to be taken first. When set to false, order of taken queue messages depends on the DBMS (usually FIFO in most cases).

Parameters:

fifo -

Returns:

markFifo

public LessLockingUniversalMySQLQueue markFifo(boolean fifo)

When set to true, queue message with lower id is ensured to be taken first. When set to false, order of taken queue messages depends on the DBMS (usually FIFO in most cases).

Parameters:

fifo -

Returns:

isFifo

public boolean isFifo()

If true, queue message with lower id is ensured to be taken first. Otherwise, order of taken queue messages depends on the DBMS (usually FIFO in most cases).

Returns:

getFifo

public boolean getFifo()

If true, queue message with lower id is ensured to be taken first. Otherwise, order of taken queue messages depends on the DBMS (usually FIFO in most cases).

Returns:

getTableNameEphemeral

public String getTableNameEphemeral()

Overrides:

getTableNameEphemeral in class JdbcQueue

init

public LessLockingUniversalMySQLQueue init()

Description copied from class: JdbcQueue

Overrides:

init in class JdbcQueue

readFromQueueStorage

protected UniversalQueueMessage readFromQueueStorage(org.springframework.jdbc.core.JdbcTemplate jdbcTemplate)

Reads a message from head of queue storage.

Specified by:

readFromQueueStorage in class JdbcQueue

Returns:

readFromEphemeralStorage

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

Reads a message from the ephemeral storage.

Specified by:

readFromEphemeralStorage in class JdbcQueue

Returns:

getOrphanFromEphemeralStorage

protected 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).

Specified by:

getOrphanFromEphemeralStorage in class JdbcQueue

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

Returns:

putToQueueStorage

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

Puts a message to tail of the queue storage.

Specified by:

putToQueueStorage in class JdbcQueue

Returns:

putToEphemeralStorage

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

Puts a message to the ephemeral storage.

Specified by:

putToEphemeralStorage in class JdbcQueue

Returns:

removeFromQueueStorage

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

Removes a message from the queue storage.

Specified by:

removeFromQueueStorage in class JdbcQueue

Returns:

removeFromEphemeralStorage

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

Removes a message from the ephemeral storage.

Specified by:

removeFromEphemeralStorage in class JdbcQueue

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.

Overrides:

_queueWithRetries in class JdbcQueue

Returns:

Throws:

SQLException

_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.

Overrides:

_requeueWithRetries in class JdbcQueue

Returns:

Throws:

SQLException

_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.

Overrides:

_requeueSilentWithRetries in class JdbcQueue

Returns:

Throws:

SQLException

_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.

Overrides:

_finishWithRetries in class JdbcQueue

Throws:

SQLException

_takeWithRetries

protected UniversalQueueMessage _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.

Overrides:

_takeWithRetries in class JdbcQueue

Returns:

Throws:

SQLException

_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.

Overrides:

_moveFromEphemeralToQueueStorageWithRetries in class JdbcQueue

Returns:

Throws:

SQLException

take

public UniversalQueueMessage 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

Overrides:

take in class JdbcQueue

Returns: