?? batchingrdbmsadapter.java
字號:
/**
* Redistribution and use of this software and associated documentation
* ("Software"), with or without modification, are permitted provided
* that the following conditions are met:
*
* 1. Redistributions of source code must retain copyright
* statements and notices. Redistributions must also contain a
* copy of this document.
*
* 2. Redistributions in binary form must reproduce the
* above copyright notice, this list of conditions and the
* following disclaimer in the documentation and/or other
* materials provided with the distribution.
*
* 3. The name "Exolab" must not be used to endorse or promote
* products derived from this Software without prior written
* permission of Exoffice Technologies. For written permission,
* please contact info@exolab.org.
*
* 4. Products derived from this Software may not be called "Exolab"
* nor may "Exolab" appear in their names without prior written
* permission of Exoffice Technologies. Exolab is a registered
* trademark of Exoffice Technologies.
*
* 5. Due credit should be given to the Exolab Project
* (http://www.exolab.org/).
*
* THIS SOFTWARE IS PROVIDED BY EXOFFICE TECHNOLOGIES AND CONTRIBUTORS
* ``AS IS'' AND ANY EXPRESSED OR IMPLIED WARRANTIES, INCLUDING, BUT
* NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND
* FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
* EXOFFICE TECHNOLOGIES OR ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
* INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
* (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
* SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
* HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
* STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED
* OF THE POSSIBILITY OF SUCH DAMAGE.
*
* Copyright 2002-2005 (C) Exoffice Technologies Inc. All Rights Reserved.
*
* $Id: BatchingRdbmsAdapter.java,v 1.2 2005/03/18 04:05:52 tanderson Exp $
*/
package org.exolab.jms.persistence;
import java.sql.Connection;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Vector;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.exolab.jms.authentication.User;
import org.exolab.jms.client.JmsDestination;
import org.exolab.jms.events.EventHandler;
import org.exolab.jms.message.MessageId;
import org.exolab.jms.message.MessageImpl;
import org.exolab.jms.messagemgr.MessageHandle;
/**
* The batching RDBMS adapter is used to improve the performance of persistent
* messages by batching more instructions into a single transaction.
* <p>
* It encpasulates an {@link RDBMSAdapter} and delegates all the behaviour to
* it. In the interim it will only batch 'insert', 'update' and 'delete'
* statements. If it receives a 'select' or any query it will first commit the
* batched statements before satisfying the query.
* <p>
* This piece of code is still under development and could effect the integrity
* of your system if used. It could also improve the throughput of persistent
* messages if it works correctly.
*
* @version $Revision: 1.2 $ $Date: 2005/03/18 04:05:52 $
* @author <a href="mailto:jima@intalio.org">Jim Alateras</a>
*/
public class BatchingRdbmsAdapter
extends PersistenceAdapter
implements EventHandler {
/**
* This is the maximum number of statements that can be batched before
* actually committing the work to the database. This value defaults to
* 100 but it can be changed at runtime.
*/
private int _maxStatementsToBatch = 500;
/**
* The directory where the log files are stored. This can be set by the
* client
*/
private String _logDirectory = ".";
/**
* Holds a reference to the RDBMSAdapter, which it delegates all the
* work too.
*/
private RDBMSAdapter _rdbms = null;
/**
* Holds the current batch of trnasactional objects.
*/
private LinkedList _batch = new LinkedList();
/**
* Holds a reference to message handles that have been batched up.
*/
private HashMap _handles = new HashMap();
/**
* Holds a reference to messages that have been batched up
*/
private HashMap _messages = new HashMap();
/**
* The logger
*/
private static final Log _log =
LogFactory.getLog(BatchingRdbmsAdapter.class);
/**
* Connects to the given db.
*
* @param driver - the rdbms driver to use
* @param url - the url to the database
* @param userName - a valid user name for a connection to the database
* @param password - the password for the connection to the database
* @param batchSize - the number of requests it should batch
* @throws PersistenceException for any database error
*/
BatchingRdbmsAdapter(String driver, String url, String userName,
String password, int batchSize)
throws PersistenceException {
// create the rdbms adapter
_rdbms = new RDBMSAdapter(driver, url, userName, password);
_maxStatementsToBatch = batchSize;
}
/**
* Close the database if open.
*
*/
public void close() {
if (_rdbms != null) {
try {
flush();
} catch (PersistenceException exception) {
_log.error("Failed to flush statements", exception);
}
_rdbms.close();
}
}
/**
* Set the maximum number of SQL statements to batch
*
* @param max - number of statements to batch
*/
public void setMaxStatementsToBatch(int max) {
_maxStatementsToBatch = max;
}
/**
* Return the maximum number of statements to batch
*
* @return int
*/
public int getMaxStatementsToBatch() {
return _maxStatementsToBatch;
}
// implementation of PersistenceAdapter.getLastId
public long getLastId(Connection connection)
throws PersistenceException {
return _rdbms.getLastId(connection);
}
// implementation of PersistenceAdapter.updateIds
public void updateIds(Connection connection, long id)
throws PersistenceException {
_rdbms.updateIds(connection, id);
}
// implementation of PersistenceMessage.addMessage
public void addMessage(Connection connection, MessageImpl message)
throws PersistenceException {
addToBatch(TransactionalObjectWrapper.ADD_MESSAGE, message);
}
// implementation of PersistenceMessage.addMessage
public void updateMessage(Connection connection, MessageImpl message)
throws PersistenceException {
addToBatch(TransactionalObjectWrapper.UPDATE_MESSAGE, message);
}
// implementation of PersistenceAdapter.getUnprocessedMessages
public Vector getUnprocessedMessages(Connection connection)
throws PersistenceException {
flush();
return _rdbms.getUnprocessedMessages(connection);
}
// implementation of PersistenceAdapter.removeMessage
public void removeMessage(Connection connection, String id)
throws PersistenceException {
addToBatch(TransactionalObjectWrapper.DELETE_MESSAGE, id);
}
// implementation of PersistenceAdapter.getMessage
public MessageImpl getMessage(Connection connection, String id)
throws PersistenceException {
flush();
return _rdbms.getMessage(connection, id);
}
// implementation of PersistenceAdapter.getMessages
public Vector getMessages(Connection connection,
MessageHandle handle)
throws PersistenceException {
flush();
return _rdbms.getMessages(connection, handle);
}
// implementation of PersistenceAdapter.addMessageHandle
public void addMessageHandle(Connection connection,
MessageHandle handle)
throws PersistenceException {
addToBatch(TransactionalObjectWrapper.ADD_HANDLE, handle);
}
// implementation of PersistenceAdapter.updateMessageHandle
public void updateMessageHandle(Connection connection,
MessageHandle handle)
throws PersistenceException {
addToBatch(TransactionalObjectWrapper.UPDATE_HANDLE, handle);
}
// implementation of PersistenceAdapter.removeMessageHandle
public void removeMessageHandle(Connection connection,
MessageHandle handle)
throws PersistenceException {
addToBatch(TransactionalObjectWrapper.DELETE_HANDLE, handle);
}
// implementation of PersistenceAdapter.getMessageHandles
public Vector getMessageHandles(Connection connection,
JmsDestination destination, String name)
throws PersistenceException {
flush();
return _rdbms.getMessageHandles(connection, destination, name);
}
// implementation of PersistenceAdapter.addDurableConsumer
public void addDurableConsumer(Connection connection, String topic,
String consumer)
throws PersistenceException {
flush();
_rdbms.addDurableConsumer(connection, topic, consumer);
}
// implementation of PersistenceAdapter.removeDurableConsumer
public void removeDurableConsumer(Connection connection, String consumer)
throws PersistenceException {
flush();
_rdbms.removeDurableConsumer(connection, consumer);
}
// implementation of PersistenceAdapter.getDurableConsumers
public Enumeration getDurableConsumers(Connection connection,
String topic)
throws PersistenceException {
flush();
return _rdbms.getDurableConsumers(connection, topic);
}
// implementation of PersistenceAdapter.getAllDurableConsumers
public HashMap getAllDurableConsumers(Connection connection)
throws PersistenceException {
flush();
return _rdbms.getAllDurableConsumers(connection);
}
// implementation of PersistenceAdapter.durableConsumerExists
public boolean durableConsumerExists(Connection connection, String name)
throws PersistenceException {
flush();
return _rdbms.durableConsumerExists(connection, name);
}
// implementation of PersistenceAdapter.addDestination
public void addDestination(Connection connection, String name,
boolean queue)
throws PersistenceException {
flush();
_rdbms.addDestination(connection, name, queue);
}
// implementation of PersistenceAdapter.removeDestination
public void removeDestination(Connection connection, String name)
throws PersistenceException {
flush();
_rdbms.removeDestination(connection, name);
}
// implementation of PersistenceAdapter.getAllDestinations
public Enumeration getAllDestinations(Connection connection)
throws PersistenceException {
flush();
return _rdbms.getAllDestinations(connection);
}
// implementation of PersistenceAdapter.checkDestination
public boolean checkDestination(Connection connection, String name)
throws PersistenceException {
flush();
return _rdbms.checkDestination(connection, name);
}
// implementation of getQueueMessageCount
public int getQueueMessageCount(Connection connection, String name)
throws PersistenceException {
flush();
return _rdbms.getQueueMessageCount(connection, name);
}
// implementation of PersistenceAdapter.getQueueMessageCount
public int getDurableConsumerMessageCount(Connection connection,
String destination, String name)
throws PersistenceException {
flush();
return _rdbms.getDurableConsumerMessageCount(connection, destination,
name);
}
// implementation of PersistenceAdapter.getQueueMessageCount
public void removeExpiredMessages(Connection connection)
throws PersistenceException {
flush();
_rdbms.removeExpiredMessages(connection);
}
// implementation of PersistenceAdapter.removeExpiredMessageHandles
public void removeExpiredMessageHandles(Connection connection,
String consumer)
throws PersistenceException {
flush();
_rdbms.removeExpiredMessageHandles(connection, consumer);
}
// implementation of PersistenceAdapter.getQueueMessageCount
public Vector getNonExpiredMessages(Connection connection,
JmsDestination destination)
throws PersistenceException {
?? 快捷鍵說明
復制代碼
Ctrl + C
搜索代碼
Ctrl + F
全屏模式
F11
切換主題
Ctrl + Shift + D
顯示快捷鍵
?
增大字號
Ctrl + =
減小字號
Ctrl + -