database.java

监控大型网络的软件。能够自动发现拓扑结构

/*
 * Database
 * 
 * $LastChangedRevision: 4221 $
 *
 * $LastChangedDate: 2007-09-24 10:02:54 +0200 (man, 24 sep 2007) $
 *
 * Copyright 2002-2004 Norwegian University of Science and Technology
 * 
 * This file is part of Network Administration Visualized (NAV)
 * 
 * NAV is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 * 
 * NAV is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with NAV; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 */

package no.ntnu.nav.Database;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.IdentityHashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import no.ntnu.nav.logger.Log;

/**
 * <p> Wrapper around the JDBC API to simplify working with the
 * database. This class contains only static methods any may thus be
 * used without needing a reference to it. It is also fully
 * thread-safe and supports per-thread transactions. </p>
 *
 * @version $LastChangedRevision: 4221 $ $LastChangedDate: 2007-09-24 10:02:54 +0200 (man, 24 sep 2007) $
 * @author Kristian Eide &lt;[e-mail]&gt;
 */

public class Database
{
	public static final long CONNECTION_IDLE_TIMEOUT = 60 * 60 * 1000; // 60 minutes

	public static final int POSTGRESQL_DRIVER = 0;
	public static final int MYSQL_DRIVER = 10;
	public static final int ORACLE_DRIVER = 20;

	public static final int DEFAULT_DRIVER = POSTGRESQL_DRIVER;
	
	public static final int DEFAULT_GLOBAL_STATEMENT_BUFFER = 32;
	public static final int DEFAULT_THREAD_STATEMENT_BUFFER = 2;
	public static final int DEFAULT_RECONNECT_WAIT_TIME = 10000;
	public static final int DEFAULT_MAX_CONNECTIONS = 6;
	public static final int DEFAULT_MAX_TRANS_CONNECTIONS = 4;

	private static final String dbDriverOracle = "oracle.jdbc.driver.OracleDriver";
	private static final String dbDriverPostgresql = "org.postgresql.Driver";
	private static final String dbDriverMysql = "org.gjt.mm.mysql.Driver";


	private static Map activeDB = Collections.synchronizedMap(new HashMap()); // Maps thread -> active connection identifier
	private static Map dbDescrs = Collections.synchronizedMap(new HashMap()); // Maps user-supplied connection identifier to DB descr

	private static Map statementMap = Collections.synchronizedMap(new IdentityHashMap()); // Used for freeing not autoclose statements

	private static class DBDescr {
		private int dbDriver;
		private String dbDriverString;
		public String serverName;
		public String serverPort;
		public String dbName;
		public String user;
		public String pw;
		private String conStr;

		private boolean returnOnReconnectFail = false;
		private int refCount;
		private LinkedList conQ;
		private LinkedList conQT;
		private Map activeTransactions = new HashMap();
		private int conCnt;
		private int conTCnt;

		private Map statementMap = Collections.synchronizedMap(new HashMap()); // Maps thread to a LinkedList of statements
		private LinkedList globStatementQ = new LinkedList();

		public DBDescr(int _dbDriver, String _serverName, String _serverPort, String _dbName, String _user, String _pw) {
			dbDriver = _dbDriver;
			serverName = _serverName;
			serverPort = _serverPort;
			dbName = _dbName;
			user = _user;
			pw = _pw;
			conQ = new LinkedList();
			conQT = new LinkedList();
			refCount = 0;
			storeConnectString();
		}

		public String getKey() {
			return conStr+":"+user+":"+pw;
		}

		public boolean verifyConnection() {
			try {
				Statement st = getStatement();
				ResultSet rs = st.executeQuery("SELECT 1");
				if (rs.next()) {
					// Connection OK
					return true;
				}
			} catch (Exception e) {
				String msg = e.getMessage();
				int idx;
				if (msg != null && (idx=msg.indexOf("Stack Trace")) >= 0) msg = msg.substring(0, idx-1);
				System.err.println("Connection verify failed: " + msg);
			}
			return false;
		}

		public int getConnectionCount() {
			return conCnt;
		}

		// Gets a connection for the current thread
		public ConnectionDescr getConnection() throws SQLException {
			synchronized (activeTransactions) {
				if (activeTransactions.containsKey(Thread.currentThread())) {
					ConnectionDescr transDescr = (ConnectionDescr)activeTransactions.get(Thread.currentThread());
					if (transDescr == null) throw new SQLException("Transaction aborted due to database reconnect");
					return transDescr;
				}
			}
			cleanConnectionQ();
			synchronized (conQ) {
				if (conQ.isEmpty()) {
					if (conCnt >= DEFAULT_MAX_CONNECTIONS) {
						waitForConnection(conQ);
					} else {
						if (!createConnection()) return null;
					}
				}
				return (ConnectionDescr)conQ.removeFirst();
			}
		}

		// Gets a transaction connection for the current thread
		public ConnectionDescr getTransConnection() throws SQLException {
			synchronized (activeTransactions) {
				if (activeTransactions.containsKey(Thread.currentThread())) {
					ConnectionDescr transDescr = (ConnectionDescr)activeTransactions.get(Thread.currentThread());
					if (transDescr == null) throw new SQLException("Transaction aborted due to database reconnect");
					return transDescr;
				}
			}
			cleanConnectionQT();
			synchronized (conQT) {
				if (conQT.isEmpty()) {
					if (conTCnt >= DEFAULT_MAX_TRANS_CONNECTIONS) {
						waitForConnection(conQT);
					} else {
						if (!createTransConnection()) return null;
					}
				}
				return (ConnectionDescr)conQT.removeFirst();
			}
		}

		public void waitForConnection(Collection waitQ) {
			// Wait for a free connection
			while (true) {
				try {
					//System.err.println("Waiting for connection...("+conQ.size()+")");
					waitQ.wait();
					//System.err.println("Got notify! ("+conQ.size()+")");
					if (!waitQ.isEmpty()) break;
				} catch (InterruptedException e) {
				}
				// There should now be a new connection available				
			}
		}

		/**
		 * <p>Free a given ConnectionDescr instance , by placing it back into 
		 * the connection queue, thus making it available for other threads to
		 * use.</p>
		 * 
		 * <p>If the given ConnectionDescr object happens to be a transaction
		 * connection for the current thread, this method does nothing.
		 * Transaction objects must be explicitly committed or rolled back 
		 * before they are freed.</p>
		 * 
		 * @param cd The ConnectionDescr object to free.
		 */
		public void freeConnection(ConnectionDescr cd) {
			synchronized (activeTransactions) {
				if (activeTransactions.get(Thread.currentThread()) == cd) {
					//System.out.println("["+Integer.toHexString(Thread.currentThread().hashCode())+"] "+"NOT freeing con");
					return;
				}
				//System.out.println("["+Integer.toHexString(Thread.currentThread().hashCode())+"] "+"FREEING con");
			}
			synchronized (conQ) {
				conQ.add(cd);
				conQ.notify();
			}
		}

		public void freeTransConnection(ConnectionDescr cd) {
			synchronized (conQT) {
				conQT.add(cd);
				conQT.notify();
			}
		}

		public Statement getStatement() throws SQLException {
			return getStatement(true);
		}

		public Statement getStatement(boolean autoclose) throws SQLException {
			ConnectionDescr cd = getConnection();
			Statement st = cd.getStatement(autoclose);
			freeConnection(cd);
			return st;
		}

		public Statement getUpdateStatement() throws SQLException {
			ConnectionDescr cd = getConnection();
			Statement st = cd.getUpdateStatement();
			/* XXX: The next line seems like a dubious scheme.  The connection 
			 * is freed and made available to other threads before the update 
			 * statement we've obtained is even used.  See further comments in 
			 * ConnectionDescr.getUpdateStatement(), called above. 
			 */
			freeConnection(cd);
			return st;
		}

		public boolean openConnection() {
			if (!connect()) return false;
			refCount++;
			return true;
		}

		private boolean connect() {
			synchronized (conQ) {
				if (conQ.isEmpty()) {
					if (!createConnection()) return false;
				}
			}
			return true;
		}

		public void closeConnection() {
			refCount--;
			if (refCount > 0) return; // Still open connections
			closeAllConnections();
		}

		public boolean reconnect() {
			synchronized (conQ) {
				//System.out.println("Closing all connections: " + conQ.size());
				closeAllConnections();
				//System.out.println("            connections: " + conQ.size());
				return connect();
			}
		}

		private void closeAllConnections() {
			synchronized (conQ) {
				closeAllConnections(conQ.iterator());
				statementMap.clear();
				globStatementQ.clear();
				conCnt = 0;
			}
			synchronized (conQT) {
				closeAllConnections(conQT.iterator());
				conTCnt = 0;
			}
			synchronized (activeTransactions) {
				List l = new ArrayList();
				for (Iterator it = activeTransactions.entrySet().iterator(); it.hasNext();) {
					Map.Entry me = (Map.Entry)it.next();
					try {
						((ConnectionDescr)me.getValue()).close();
					} catch (SQLException e) {
						String msg = e.getMessage();
						int idx;
						if (msg != null && (idx=msg.indexOf("Stack Trace")) >= 0) msg = msg.substring(0, idx-1);
						System.err.println("closeConnection error: " + msg);
					}
					l.add(me.getKey());
				}
				for (Iterator it = l.iterator(); it.hasNext();) {
					activeTransactions.put(it.next(), null);
				}
			}
		}

		private void closeAllConnections(Iterator conIt) {
			while (conIt.hasNext()) {
				try {
					((ConnectionDescr)conIt.next()).close();
				} catch (SQLException e) {
					String msg = e.getMessage();
					int idx;
					if (msg != null && (idx=msg.indexOf("Stack Trace")) >= 0) msg = msg.substring(0, idx-1);
					System.err.println("closeConnection error: " + msg);
				}
				conIt.remove();
			}
		}

		private void cleanConnectionQ() throws SQLException {
			synchronized (conQ) {
				conCnt -= cleanConnectionQ(conQ);
			}
		}

		private void cleanConnectionQT() throws SQLException {
			synchronized (conQT) {
				conTCnt -= cleanConnectionQ(conQT);				
			}			
		}

		private int cleanConnectionQ(LinkedList conQ) throws SQLException {
			int remCnt = 0;
			if (conQ.size() > 1) {
				Iterator it = conQ.iterator();
				while (it.hasNext()) {
					ConnectionDescr cd = (ConnectionDescr)it.next();
					if (System.currentTimeMillis() - cd.lastUsed > CONNECTION_IDLE_TIMEOUT) {
						cd.close();
						it.remove();
						remCnt++;
						if (conQ.size() == 1) break;
					}
				}
			}
			return remCnt;
		}

		private boolean createConnection() {
			ConnectionDescr cd = newConnection();
			if (cd != null) {
				synchronized (conQ) {
					conQ.add(cd);
					conCnt++;
					conQ.notify();
					return true;
				}
			}
			return false;
		}

		private boolean createTransConnection() {
			ConnectionDescr cd = newConnection();
			if (cd != null) {
				synchronized (conQT) {
					conQT.add(cd);
					conTCnt++;
					conQT.notify();