MySQL驱动阅读------Connection连接的建立，基于JDBC-----5.1.26

Posted 2020-06-09

 

一般获取数据库连接的程序

Class.forName("com.mysql.jdbc.Driver");
final Connection connection = （Connection）DriverManager.getConnection("jdbc:mysql://localhost:3306/testDB","guoxiaoming","guoxiaoming");

-----------------------------

注册mysql驱动类下次再说，目前重点介绍Connection的建立

-----------------------------

public interface Connection  extends Wrapper {

JDK的Connection提供了一种标准，与特定数据库的连接（会话）。在连接上下文中执行 SQL 语句并返回结果。

该Connection接口被数据库驱动类实现，保存有数据库连接中的IO类。

JDBC当中的DriverManager类封装了获取Connection对象的统一方法

public static Connection getConnection(String url, String user, String password) throws SQLException {
        java.util.Properties info = new java.util.Properties();

        // Gets the classloader of the code that called this method, may 
	// be null.
	ClassLoader callerCL = DriverManager.getCallerClassLoader();

	if (user != null) {
	    info.put("user", user);
	}
	if (password != null) {
	    info.put("password", password);
	}

        return (getConnection(url, info, callerCL));
}

 

进入getConnection（url，info，callerCL）方法查看可以看到，

Connection result = di.driver.connect(url, info);

代码最终获取了Connection对象

对应的driver驱动类实现，在jdbc当中

public class NonRegisteringReplicationDriver extends NonRegisteringDriver {
	public NonRegisteringReplicationDriver() throws SQLException {
		super();
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see java.sql.Driver#connect(java.lang.String, java.util.Properties)
	 */
	public Connection connect(String url, Properties info) throws SQLException {
		return connectReplicationConnection(url, info);
	}
}

看到connectReplicationConnection(url,info)方法在父类NonRegisteringDriver类当中

return new ReplicationConnection(masterProps, slavesProps);

看到该类

public ReplicationConnection(Properties masterProperties,
			Properties slaveProperties) throws SQLException {
		this.driver = new NonRegisteringDriver();
		this.slaveProperties = slaveProperties;
		this.masterProperties = masterProperties;

        this.initializeMasterConnection();
        this.initializeSlaveConnection();
        
		this.currentConnection = this.masterConnection;
	}

初始化了两个链接，一个是masterConnection 一个是slaveConnection

看注释

/**
 * Connection that opens two connections, one two a replication master, and
 * another to one or more slaves, and decides to use master when the connection
 * is not read-only, and use slave(s) when the connection is read-only.
 * 
 * @version $Id: ReplicationConnection.java,v 1.1.2.1 2005/05/13 18:58:38
 *          mmatthews Exp $
 */

我们最终看到在ConnectionImpl类当中

protected static Connection getInstance(String hostToConnectTo,
			int portToConnectTo, Properties info, String databaseToConnectTo,
			String url) throws SQLException {
		if (!Util.isJdbc4()) {
			return new ConnectionImpl(hostToConnectTo, portToConnectTo, info,
					databaseToConnectTo, url);
		}

		return (Connection) Util.handleNewInstance(JDBC_4_CONNECTION_CTOR,
				new Object[] {
							hostToConnectTo, Integer.valueOf(portToConnectTo), info,
							databaseToConnectTo, url }, null);
	}

　　实例化了该类的一个子类

public class JDBC4Connection extends ConnectionImpl 

　通过构造方法

public JDBC4Connection(String hostToConnectTo, int portToConnectTo, Properties info, String databaseToConnectTo, String url) throws SQLException {
	super(hostToConnectTo, portToConnectTo, info, databaseToConnectTo, url);
	// TODO Auto-generated constructor stub
}

在其父类中

protected ConnectionImpl(String hostToConnectTo, int portToConnectTo, Properties info,
			String databaseToConnectTo, String url)
			throws SQLException {
	
		this.connectionCreationTimeMillis = System.currentTimeMillis();
		
		if (databaseToConnectTo == null) {
			databaseToConnectTo = "";
		}

重点关于方法

public void createNewIO(boolean isForReconnect)
			throws SQLException {
		synchronized (getConnectionMutex()) {
			// Synchronization Not needed for *new* connections, but defintely for
			// connections going through fail-over, since we might get the
			// new connection up and running *enough* to start sending
			// cached or still-open server-side prepared statements over
			// to the backend before we get a chance to re-prepare them...
			
	
			Properties mergedProps  = exposeAsProperties(this.props);
	
			if (!getHighAvailability()) {
				connectOneTryOnly(isForReconnect, mergedProps);
				
				return;
			} 
	
			connectWithRetries(isForReconnect, mergedProps);
		}		
	}

 

最终我们看到在coreConnect方法中实现了具体的链接

private void coreConnect(Properties mergedProps) throws SQLException,
			IOException {
		int newPort = 3306;
		String newHost = "localhost";
		
		String protocol = mergedProps.getProperty(NonRegisteringDriver.PROTOCOL_PROPERTY_KEY);
		
		if (protocol != null) {
			// "new" style URL

			if ("tcp".equalsIgnoreCase(protocol)) {
				newHost = normalizeHost(mergedProps.getProperty(NonRegisteringDriver.HOST_PROPERTY_KEY));
				newPort = parsePortNumber(mergedProps.getProperty(NonRegisteringDriver.PORT_PROPERTY_KEY, "3306"));
			} else if ("pipe".equalsIgnoreCase(protocol)) {
				setSocketFactoryClassName(NamedPipeSocketFactory.class.getName());
				
				String path = mergedProps.getProperty(NonRegisteringDriver.PATH_PROPERTY_KEY);
				
				if (path != null) {
					mergedProps.setProperty(NamedPipeSocketFactory.NAMED_PIPE_PROP_NAME, path);
				}
			} else {
				// normalize for all unknown protocols
				newHost = normalizeHost(mergedProps.getProperty(NonRegisteringDriver.HOST_PROPERTY_KEY));
				newPort = parsePortNumber(mergedProps.getProperty(NonRegisteringDriver.PORT_PROPERTY_KEY, "3306"));
			}
		} else {
		
			String[] parsedHostPortPair = NonRegisteringDriver
					.parseHostPortPair(this.hostPortPair);
			newHost = parsedHostPortPair[NonRegisteringDriver.HOST_NAME_INDEX];

			newHost = normalizeHost(newHost);

			if (parsedHostPortPair[NonRegisteringDriver.PORT_NUMBER_INDEX] != null) {
				newPort = parsePortNumber(parsedHostPortPair[NonRegisteringDriver.PORT_NUMBER_INDEX]);
			}
		}

		this.port = newPort;
		this.host = newHost;
		
		this.io = new MysqlIO(newHost, newPort,
				mergedProps, getSocketFactoryClassName(),
				getProxy(), getSocketTimeout(),
				this.largeRowSizeThreshold.getValueAsInt());
		this.io.doHandshake(this.user, this.password,
				this.database);
	}

 

在new MysqlIO类当中的构造方法中实现

 public MysqlIO(String host, int port, Properties props,
        String socketFactoryClassName, MySQLConnection conn,
        int socketTimeout, int useBufferRowSizeThreshold) throws IOException, SQLException {
        this.connection = conn;
        
        if (this.connection.getEnablePacketDebug()) {
            this.packetDebugRingBuffer = new LinkedList<StringBuffer>();
        }
        this.traceProtocol = this.connection.getTraceProtocol();
        

        this.useAutoSlowLog = this.connection.getAutoSlowLog();
        
        this.useBufferRowSizeThreshold = useBufferRowSizeThreshold;
        this.useDirectRowUnpack = this.connection.getUseDirectRowUnpack();

        this.logSlowQueries = this.connection.getLogSlowQueries();

        this.reusablePacket = new Buffer(INITIAL_PACKET_SIZE);
        this.sendPacket = new Buffer(INITIAL_PACKET_SIZE);

        this.port = port;
        this.host = host;

        this.socketFactoryClassName = socketFactoryClassName;
        this.socketFactory = createSocketFactory();
        this.exceptionInterceptor = this.connection.getExceptionInterceptor();
        
        try {
        	this.mysqlConnection = this.socketFactory.connect(this.host,
        		this.port, props);
	        
	
	        if (socketTimeout != 0) {
	        	try {
	        		this.mysqlConnection.setSoTimeout(socketTimeout);
	        	} catch (Exception ex) {
	        		/* Ignore if the platform does not support it */
	        	}
	        }
	
	        this.mysqlConnection = this.socketFactory.beforeHandshake();
	
	        if (this.connection.getUseReadAheadInput()) {
	        	this.mysqlInput = new ReadAheadInputStream(this.mysqlConnection.getInputStream(), 16384,
	        			this.connection.getTraceProtocol(),
	        			this.connection.getLog());
	        } else if (this.connection.useUnbufferedInput()) {
	        	this.mysqlInput = this.mysqlConnection.getInputStream();
	        } else {
	        	this.mysqlInput = new BufferedInputStream(this.mysqlConnection.getInputStream(),
	        			16384);
	        }
	
	        this.mysqlOutput = new BufferedOutputStream(this.mysqlConnection.getOutputStream(),
	        		16384);
	
	
	        this.isInteractiveClient = this.connection.getInteractiveClient();
	        this.profileSql = this.connection.getProfileSql();
	        this.autoGenerateTestcaseScript = this.connection.getAutoGenerateTestcaseScript();
	
	        this.needToGrabQueryFromPacket = (this.profileSql ||
	        		this.logSlowQueries ||
	        		this.autoGenerateTestcaseScript);
	
	        if (this.connection.getUseNanosForElapsedTime()
					&& Util.nanoTimeAvailable()) {
				this.useNanosForElapsedTime = true;
	
				this.queryTimingUnits = Messages.getString("Nanoseconds");
			} else {
				this.queryTimingUnits = Messages.getString("Milliseconds");
			}
	
			if (this.connection.getLogSlowQueries()) {
				calculateSlowQueryThreshold();
			}
        } catch (IOException ioEx) {
        	throw SQLError.createCommunicationsException(this.connection, 0, 0, ioEx, getExceptionInterceptor());
        }
    }

 

this.mysqlConnection = this.socketFactory.connect(this.host,this.port, props);
这样就返回了Socket对象

在StandardSocketFactory类当中

if (this.host != null) {
				if (!(wantsLocalBind || wantsTimeout || needsConfigurationBeforeConnect)) {
					InetAddress[] possibleAddresses = InetAddress
							.getAllByName(this.host);

					Throwable caughtWhileConnecting = null;

					// Need to loop through all possible addresses, in case
					// someone has IPV6 configured (SuSE, for example...)

					for (int i = 0; i < possibleAddresses.length; i++) {
						try {
							this.rawSocket = new Socket(possibleAddresses[i],
									port);

							configureSocket(this.rawSocket, props);

							break;
						} catch (Exception ex) {
							caughtWhileConnecting = ex;
						}
					}

					if (rawSocket == null) {
						unwrapExceptionToProperClassAndThrowIt(caughtWhileConnecting);
					}
				} else {
					// must explicitly state this due to classloader issues
					// when running on older JVMs :(
					try {

						InetAddress[] possibleAddresses = InetAddress
								.getAllByName(this.host);

						Throwable caughtWhileConnecting = null;

						Object localSockAddr = null;

						Class<?> inetSocketAddressClass = null;

						Constructor<?> addrConstructor = null;

						try {
							inetSocketAddressClass = Class
									.forName("java.net.InetSocketAddress");

							addrConstructor = inetSocketAddressClass
									.getConstructor(new Class[] {
											InetAddress.class, Integer.TYPE });

							if (wantsLocalBind) {
								localSockAddr = addrConstructor
										.newInstance(new Object[] {
												InetAddress
														.getByName(localSocketHostname),
												new Integer(0 /*
																 * use ephemeral
																 * port
																 */) });

							}
						} catch (Throwable ex) {
							unwrapExceptionToProperClassAndThrowIt(ex);
						}

						// Need to loop through all possible addresses, in case
						// someone has IPV6 configured (SuSE, for example...)

						for (int i = 0; i < possibleAddresses.length; i++) {

							try {
								this.rawSocket = new Socket();

								configureSocket(this.rawSocket, props);

								Object sockAddr = addrConstructor
										.newInstance(new Object[] {
												possibleAddresses[i],
												Integer.valueOf(port) });
								// bind to the local port if not using the ephemeral port
								if (localSockAddr != null) {
									socketBindMethod.invoke(rawSocket,
											new Object[] { localSockAddr });
								}

								connectWithTimeoutMethod.invoke(rawSocket,
										new Object[] { sockAddr,
												Integer.valueOf(connectTimeout) });

								break;
							} catch (Exception ex) {	
								this.rawSocket = null;

								caughtWhileConnecting = ex;
							}
						}

						if (this.rawSocket == null) {
							unwrapExceptionToProperClassAndThrowIt(caughtWhileConnecting);
						}

					} catch (Throwable t) {
						unwrapExceptionToProperClassAndThrowIt(t);
					}
				}

				return this.rawSocket;
			}

针对host可能对应多个ip的可能性，连接到其中一个ip之后就break退出