c3p0pooledconnectionpool.java

c3p0数据库连接池实现源码

/*
 * Distributed as part of c3p0 v.0.9.1-pre6
 *
 * Copyright (C) 2005 Machinery For Change, Inc.
 *
 * Author: Steve Waldman <[e-mail]>
 *
 * This library is free software; you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License version 2.1, as 
 * published by the Free Software Foundation.
 *
 * This software 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 Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this software; see the file LICENSE.  If not, write to the
 * Free Software Foundation, Inc., 59 Temple Place, Suite 330,
 * Boston, MA 02111-1307, USA.
 */


package com.mchange.v2.c3p0.impl;

import com.mchange.v2.c3p0.stmt.*;

import java.sql.Connection;
import java.sql.SQLException;

import javax.sql.ConnectionEvent;
import javax.sql.ConnectionEventListener;
import javax.sql.ConnectionPoolDataSource;
import javax.sql.PooledConnection;

import com.mchange.v1.db.sql.ConnectionUtils;
import com.mchange.v2.async.AsynchronousRunner;
import com.mchange.v2.async.ThreadPoolAsynchronousRunner;
import com.mchange.v2.holders.SynchronizedIntHolder;
import com.mchange.v2.log.MLevel;
import com.mchange.v2.log.MLog;
import com.mchange.v2.log.MLogger;
import com.mchange.v2.c3p0.ConnectionTester;
import com.mchange.v2.c3p0.QueryConnectionTester;
import com.mchange.v2.resourcepool.CannotAcquireResourceException;
import com.mchange.v2.resourcepool.ResourcePool;
import com.mchange.v2.resourcepool.ResourcePoolException;
import com.mchange.v2.resourcepool.ResourcePoolFactory;
import com.mchange.v2.resourcepool.TimeoutException;
import com.mchange.v2.sql.SqlUtils;

public final class C3P0PooledConnectionPool
{
    private final static boolean ASYNCHRONOUS_CONNECTION_EVENT_LISTENER = false;

    final static MLogger logger = MLog.getLogger( C3P0PooledConnectionPool.class );

    ResourcePool rp;
    ConnectionEventListener cl = new ConnectionEventListenerImpl();

    ConnectionTester     connectionTester;
    GooGooStatementCache scache;

    int checkoutTimeout;

    AsynchronousRunner sharedTaskRunner;

    C3P0PooledConnectionPool( final ConnectionPoolDataSource cpds,
			      final DbAuth auth,
			      int  min, 
			      int  max, 
			      int  inc,
			      int acq_retry_attempts,
			      int acq_retry_delay,
			      boolean break_after_acq_failure,
			      int checkoutTimeout, //milliseconds
			      int idleConnectionTestPeriod, //seconds
			      int maxIdleTime, //seconds
			      final boolean testConnectionOnCheckout,
			      final boolean testConnectionOnCheckin,
			      int maxStatements,
			      int maxStatementsPerConnection,
			      final ConnectionTester connectionTester,
			      final String testQuery,
			      final ResourcePoolFactory fact,
			      ThreadPoolAsynchronousRunner taskRunner) throws SQLException
    {
        try
	    {
		if (maxStatements > 0 && maxStatementsPerConnection > 0)
		    this.scache = new DoubleMaxStatementCache( taskRunner, maxStatements, maxStatementsPerConnection );
		else if (maxStatementsPerConnection > 0)
		    this.scache = new PerConnectionMaxOnlyStatementCache( taskRunner, maxStatementsPerConnection );
		else if (maxStatements > 0)
		    this.scache = new GlobalMaxOnlyStatementCache( taskRunner, maxStatements );
		else
		    this.scache = null;

		this.connectionTester = connectionTester;

		this.checkoutTimeout = checkoutTimeout;

		this.sharedTaskRunner = taskRunner;

		class PooledConnectionResourcePoolManager implements ResourcePool.Manager
		{	
		    SynchronizedIntHolder totalOpenedCounter  = new SynchronizedIntHolder();
		    SynchronizedIntHolder connectionCounter   = new SynchronizedIntHolder();
		    SynchronizedIntHolder failedCloseCounter  = new SynchronizedIntHolder();
		    
		    public Object acquireResource() throws Exception
		    { 
			PooledConnection out = (auth.equals( C3P0ImplUtils.NULL_AUTH ) ?
						cpds.getPooledConnection() :
						cpds.getPooledConnection( auth.getUser(), 
									  auth.getPassword() ) );
			
			connectionCounter.increment(); 
			totalOpenedCounter.increment();
			
			try
			    {
				if (scache != null)
				    {
					if (out instanceof C3P0PooledConnection)
					    ((C3P0PooledConnection) out).initStatementCache(scache);
					else if (out instanceof NewPooledConnection)
					    ((NewPooledConnection) out).initStatementCache(scache);
					else
					    {
						// System.err.print("Warning! StatementPooling not ");
						// System.err.print("implemented for external (non-c3p0) ");
						// System.err.println("ConnectionPoolDataSources.");
						
						logger.warning("StatementPooling not " +
							       "implemented for external (non-c3p0) " +
							       "ConnectionPoolDataSources.");
					    }
				    }
				out.addConnectionEventListener( cl );
				return out;
			    }
			catch (Exception e)
			    {
				if (logger.isLoggable( MLevel.WARNING ))
				    logger.warning("A PooledConnection was acquired, but an Exception occurred while preparing it for use. " +
						   "Attempting to destroy.");
				try { destroyResource( out ); }
				catch (Exception e2)
				    {
					if (logger.isLoggable( MLevel.WARNING ))
					    logger.log( MLevel.WARNING, 
							"An Exception occurred while trying to close partially acquired PooledConnection.",
							e2 );
				    }
				
				throw e;
			    }
			finally
			    {
				if (logger.isLoggable( MLevel.FINEST ))
				    logger.finest(this + ".acquireResource() returning. " +
						  "Currently open Connections: " + connectionCounter.getValue() +
						  "; Failed close count: " + failedCloseCounter.getValue() +
						  "; Total processed by this pool: " + totalOpenedCounter.getValue());
			    }
		    }
		    
		    // REFURBISHMENT:
		    // the PooledConnection refurbishes itself when 
		    // its Connection view is closed, prior to being
		    // checked back in to the pool. But we still may want to
		    // test to make sure it is still good.
		    
		    public void refurbishResourceOnCheckout( Object resc ) throws Exception
		    {
			if ( testConnectionOnCheckout )
			    {
				if ( logger.isLoggable( MLevel.FINER ) )
				    finerLoggingTestPooledConnection( resc, "CHECKOUT" );
				else
				    testPooledConnection( resc );
			    }
		    }
		    
		    public void refurbishResourceOnCheckin( Object resc ) throws Exception
		    {
			if ( testConnectionOnCheckin )
			    { 
				if ( logger.isLoggable( MLevel.FINER ) )
				    finerLoggingTestPooledConnection( resc, "CHECKIN" );
				else
				    testPooledConnection( resc );
			    }
		    }
		    
		    public void refurbishIdleResource( Object resc ) throws Exception
		    { 
			if ( logger.isLoggable( MLevel.FINER ) )
			    finerLoggingTestPooledConnection( resc, "IDLE CHECK" );
			else
			    testPooledConnection( resc );
		    }
		    
		    private void finerLoggingTestPooledConnection(Object resc, String testImpetus) throws Exception
		    {
			logger.finer("Testing PooledConnection [" + resc + "] on " + testImpetus + ".");
			try
			    {
				testPooledConnection( resc );
				logger.finer("Test of PooledConnection [" + resc + "] on "+testImpetus+" has SUCCEEDED.");
			    }
			catch (Exception e)
			    {
				logger.log(MLevel.FINER, "Test of PooledConnection [" + resc + "] on "+testImpetus+" has FAILED.", e);
				e.fillInStackTrace();
				throw e;
			    }
		    }
		    
		    private void testPooledConnection(Object resc) throws Exception
		    { 
			PooledConnection pc = (PooledConnection) resc;
			
			int status;
			Connection conn = null;
			Throwable rootCause = null;
			try	
			    { 
				//we don't want any callbacks while we're testing the resource
				pc.removeConnectionEventListener( cl );
				
				conn = pc.getConnection(); //checkout proxy connection
				if ( testQuery == null )
				    status = connectionTester.activeCheckConnection( conn );
				else
				    {
					if (connectionTester instanceof QueryConnectionTester)
					    status = ((QueryConnectionTester) connectionTester).activeCheckConnection( conn, testQuery );
					else
					    {
						// System.err.println("[c3p0] WARNING: testQuery '" + testQuery +
						// "' ignored. Please set a ConnectionTester that implements " +
						// "com.mchange.v2.c3p0.advanced.QueryConnectionTester, or use the " +
						// "DefaultConnectionTester, to test with the testQuery.");

						logger.warning("[c3p0] testQuery '" + testQuery +
							       "' ignored. Please set a ConnectionTester that implements " +
							       "com.mchange.v2.c3p0.advanced.QueryConnectionTester, or use the " +
							       "DefaultConnectionTester, to test with the testQuery.");
						status = connectionTester.activeCheckConnection( conn );
					    }
				    }
			    }
			catch (SQLException e)
			    {
				if (Debug.DEBUG)
				    logger.log(MLevel.FINE, "A Connection test failed with an Exception.", e);
				//e.printStackTrace();
				status = ConnectionTester.CONNECTION_IS_INVALID;
// 				System.err.println("rootCause ------>");
// 				e.printStackTrace();
				rootCause = e;
			    }
			finally
			    { 
				ConnectionUtils.attemptClose( conn ); //invalidate proxy connection
				pc.addConnectionEventListener( cl );  //should we move this to CONNECTION_IS_OKAY case? (it should work either way)
			    }
			
			switch (status)
			    {
			    case ConnectionTester.CONNECTION_IS_OKAY:
				break; //no problem, babe
			    case ConnectionTester.DATABASE_IS_INVALID:
				rp.resetPool();
				//intentional cascade...
			    case ConnectionTester.CONNECTION_IS_INVALID:
				Exception throwMe;
				if (rootCause == null)
				    throwMe = new SQLException("Connection is invalid");
				else
				    throwMe = SqlUtils.toSQLException("Connection is invalid", rootCause);
				throw throwMe;
			    default:
				throw new Error("Bad Connection Tester (" + 
						connectionTester + ") " +
						"returned invalid status (" + status + ").");
			    }
		    }
			
		    public void destroyResource(Object resc) throws Exception