JDBC源码——MySQL篇

注册驱动

看似相当不起眼的一个操作，仅仅只是注册了下Mysql的Driver实现，实则内部隐含大秘密。在派生关系中，有一个NonRegisteringDriver类，是com.mysql.cj.jdbc.Driver的父类，在其静态代码块中，开启了回收丢弃连接的自旋任务，详情见com.mysql.cj.jdbc.AbandonedConnectionCleanupThread#AbandonedConnectionCleanupThread。

static {
        try {
            Class.forName(AbandonedConnectionCleanupThread.class.getName());
        } catch (ClassNotFoundException e) {
            // ignore
        }
    }

public void run() {
        for (;;) {
            try {
                checkThreadContextClassLoader();
                Reference<? extends MysqlConnection> reference = referenceQueue.remove(5000);
                if (reference != null) {
                    finalizeResource((ConnectionFinalizerPhantomReference) reference);
                }
            } catch (InterruptedException e) {
                threadRefLock.lock();
                try {
                    threadRef = null;
                    // Finalize remaining references.
                    Reference<? extends MysqlConnection> reference;
                    while ((reference = referenceQueue.poll()) != null) {
                        finalizeResource((ConnectionFinalizerPhantomReference) reference);
                    }
                    connectionFinalizerPhantomRefs.clear();
                } finally {
                    threadRefLock.unlock();
                }
                return;
            } catch (Exception ex) {
                // Nowhere to really log this.
            }
        }
    }

打开连接

java.sql.DriverManager#getConnection(java.lang.String, java.lang.String, java.lang.String)通过委托模式，最终会调用到MySQL驱动包的connect方法
1、判断url是否支持
2、拼接参数获取cacheKey
3、从缓存中获取，如果缓存没有则生成url解析器，采用了读写锁做并发控制
4、调用com.mysql.cj.jdbc.ConnectionImpl#getInstance
5、调用com.mysql.cj.jdbc.ConnectionImpl#ConnectionImpl(com.mysql.cj.conf.HostInfo)
6、通过入参获取连接配置信息，ip，port，username，password
7、com.mysql.cj.jdbc.ConnectionImpl#createNewIO则是进行socket的建立（线程安全）
8、通过前面创建的session对象进行socketFactory的创建，以及连接socket
9、socket通信前握手

public ConnectionImpl(HostInfo hostInfo) throws SQLException {
        try {
            // Stash away for later, used to clone this connection for Statement.cancel and Statement.setQueryTimeout().
            this.origHostInfo = hostInfo;
            this.origHostToConnectTo = hostInfo.getHost();
            this.origPortToConnectTo = hostInfo.getPort();
            this.database = hostInfo.getDatabase();
            this.user = StringUtils.isNullOrEmpty(hostInfo.getUser()) ? "" : hostInfo.getUser();
            this.password = StringUtils.isNullOrEmpty(hostInfo.getPassword()) ? "" : hostInfo.getPassword();
            this.props = hostInfo.exposeAsProperties();
            this.propertySet = new JdbcPropertySetImpl();
            this.propertySet.initializeProperties(this.props);
            // We need Session ASAP to get access to central driver functionality
            this.nullStatementResultSetFactory = new ResultSetFactory(this, null);
            this.session = new NativeSession(hostInfo, this.propertySet);
            this.session.addListener(this); // listen for session status changes
            // we can't cache fixed values here because properties are still not initialized with user provided values
            this.autoReconnectForPools = this.propertySet.getBooleanProperty(PropertyKey.autoReconnectForPools);
            this.cachePrepStmts = this.propertySet.getBooleanProperty(PropertyKey.cachePrepStmts);
            this.autoReconnect = this.propertySet.getBooleanProperty(PropertyKey.autoReconnect);
            this.useUsageAdvisor = this.propertySet.getBooleanProperty(PropertyKey.useUsageAdvisor);
            this.reconnectAtTxEnd = this.propertySet.getBooleanProperty(PropertyKey.reconnectAtTxEnd);
            this.emulateUnsupportedPstmts = this.propertySet.getBooleanProperty(PropertyKey.emulateUnsupportedPstmts);
            this.ignoreNonTxTables = this.propertySet.getBooleanProperty(PropertyKey.ignoreNonTxTables);
            this.pedantic = this.propertySet.getBooleanProperty(PropertyKey.pedantic);
            this.prepStmtCacheSqlLimit = this.propertySet.getIntegerProperty(PropertyKey.prepStmtCacheSqlLimit);
            this.useLocalSessionState = this.propertySet.getBooleanProperty(PropertyKey.useLocalSessionState);
            this.useServerPrepStmts = this.propertySet.getBooleanProperty(PropertyKey.useServerPrepStmts);
            this.processEscapeCodesForPrepStmts = this.propertySet.getBooleanProperty(PropertyKey.processEscapeCodesForPrepStmts);
            this.useLocalTransactionState = this.propertySet.getBooleanProperty(PropertyKey.useLocalTransactionState);
            this.disconnectOnExpiredPasswords = this.propertySet.getBooleanProperty(PropertyKey.disconnectOnExpiredPasswords);
            this.readOnlyPropagatesToServer = this.propertySet.getBooleanProperty(PropertyKey.readOnlyPropagatesToServer);
            String exceptionInterceptorClasses = this.propertySet.getStringProperty(PropertyKey.exceptionInterceptors).getStringValue();
            if (exceptionInterceptorClasses != null && !"".equals(exceptionInterceptorClasses)) {
                this.exceptionInterceptor = new ExceptionInterceptorChain(exceptionInterceptorClasses, this.props, this.session.getLog());
            }
            if (this.cachePrepStmts.getValue()) {
                createPreparedStatementCaches();
            }
            if (this.propertySet.getBooleanProperty(PropertyKey.cacheCallableStmts).getValue()) {
                this.parsedCallableStatementCache = new LRUCache<>(this.propertySet.getIntegerProperty(PropertyKey.callableStmtCacheSize).getValue());
            }
            if (this.propertySet.getBooleanProperty(PropertyKey.allowMultiQueries).getValue()) {
                this.propertySet.getProperty(PropertyKey.cacheResultSetMetadata).setValue(false); // we don't handle this yet
            }
            if (this.propertySet.getBooleanProperty(PropertyKey.cacheResultSetMetadata).getValue()) {
                this.resultSetMetadataCache = new LRUCache<>(this.propertySet.getIntegerProperty(PropertyKey.metadataCacheSize).getValue());
            }
            if (this.propertySet.getStringProperty(PropertyKey.socksProxyHost).getStringValue() != null) {
                this.propertySet.getProperty(PropertyKey.socketFactory).setValue(SocksProxySocketFactory.class.getName());
            }
            this.pointOfOrigin = this.useUsageAdvisor.getValue() ? LogUtils.findCallingClassAndMethod(new Throwable()) : "";
            this.dbmd = getMetaData(false, false);
            initializeSafeQueryInterceptors();
        } catch (CJException e1) {
            throw SQLExceptionsMapping.translateException(e1, getExceptionInterceptor());
        }
        try {
            createNewIO(false);
            unSafeQueryInterceptors();
            AbandonedConnectionCleanupThread.trackConnection(this, this.getSession().getNetworkResources());
        } catch (SQLException ex) {
            cleanup(ex);
            // don't clobber SQL exceptions
            throw ex;
        } catch (Exception ex) {
            cleanup(ex);
            throw SQLError
                    .createSQLException(
                            this.propertySet.getBooleanProperty(PropertyKey.paranoid).getValue() ? Messages.getString("Connection.0")
                                    : Messages.getString("Connection.1",
                                            new Object[] { this.session.getHostInfo().getHost(), this.session.getHostInfo().getPort() }),
                            MysqlErrorNumbers.SQL_STATE_COMMUNICATION_LINK_FAILURE, ex, getExceptionInterceptor());
        }
    }

private void connectOneTryOnly(boolean isForReconnect) throws SQLException {
        Exception connectionNotEstablishedBecause = null;
        try {
            JdbcConnection c = getProxy();
            this.session.connect(this.origHostInfo, this.user, this.password, this.database, DriverManager.getLoginTimeout() * 1000, c);
            // save state from old connection
            boolean oldAutoCommit = getAutoCommit();
            int oldIsolationLevel = this.isolationLevel;
            boolean oldReadOnly = isReadOnly(false);
            String oldCatalog = getCatalog();
            this.session.setQueryInterceptors(this.queryInterceptors);
            // Server properties might be different from previous connection, so initialize again...
            initializePropsFromServer();
            if (isForReconnect) {
                // Restore state from old connection
                setAutoCommit(oldAutoCommit);
                setTransactionIsolation(oldIsolationLevel);
                setCatalog(oldCatalog);
                setReadOnly(oldReadOnly);
            }
            return;
        } catch (UnableToConnectException rejEx) {
            close();
            this.session.getProtocol().getSocketConnection().forceClose();
            throw rejEx;
        } catch (Exception EEE) {
            if ((EEE instanceof PasswordExpiredException
                    || EEE instanceof SQLException && ((SQLException) EEE).getErrorCode() == MysqlErrorNumbers.ER_MUST_CHANGE_PASSWORD)
                    && !this.disconnectOnExpiredPasswords.getValue()) {
                return;
            }
            if (this.session != null) {
                this.session.forceClose();
            }
            connectionNotEstablishedBecause = EEE;
            if (EEE instanceof SQLException) {
                throw (SQLException) EEE;
            }
            if (EEE.getCause() != null && EEE.getCause() instanceof SQLException) {
                throw (SQLException) EEE.getCause();
            }
            if (EEE instanceof CJException) {
                throw (CJException) EEE;
            }
            SQLException chainedEx = SQLError.createSQLException(Messages.getString("Connection.UnableToConnect"),
                    MysqlErrorNumbers.SQL_STATE_UNABLE_TO_CONNECT_TO_DATASOURCE, getExceptionInterceptor());
            chainedEx.initCause(connectionNotEstablishedBecause);
            throw chainedEx;
        }
    }

构建statement

会将先前创建的连接，会话等对象作为成员变量

执行SQL

1、检查语句是否为空等基础校验，另外可能还会进行转义操作
2、sendQueryString发送SQL语句并接收返回结果集
3、NativeProtocol进行结果的读取

public final <T extends Resultset> T sendQueryString(Query callingQuery, String query, String characterEncoding, int maxRows, boolean streamResults,
            String catalog, ColumnDefinition cachedMetadata, GetProfilerEventHandlerInstanceFunction getProfilerEventHandlerInstanceFunction,
            ProtocolEntityFactory<T, NativePacketPayload> resultSetFactory) throws IOException {
        String statementComment = this.queryComment;
        if (this.propertySet.getBooleanProperty(PropertyKey.includeThreadNamesAsStatementComment).getValue()) {
            statementComment = (statementComment != null ? statementComment + ", " : "") + "java thread: " + Thread.currentThread().getName();
        }
        // We don't know exactly how many bytes we're going to get from the query. Since we're dealing with UTF-8, the max is 4, so pad it
        // (4 * query) + space for headers
        int packLength = 1 + (query.length() * 4) + 2;
        byte[] commentAsBytes = null;
        if (statementComment != null) {
            commentAsBytes = StringUtils.getBytes(statementComment, characterEncoding);
            packLength += commentAsBytes.length;
            packLength += 6; // for /*[space] [space]*/
        }
        // TODO decide how to safely use the shared this.sendPacket
        //if (this.sendPacket == null) {
        NativePacketPayload sendPacket = new NativePacketPayload(packLength);
        //}
        sendPacket.setPosition(0);
        sendPacket.writeInteger(IntegerDataType.INT1, NativeConstants.COM_QUERY);
        if (commentAsBytes != null) {
            sendPacket.writeBytes(StringLengthDataType.STRING_FIXED, Constants.SLASH_STAR_SPACE_AS_BYTES);
            sendPacket.writeBytes(StringLengthDataType.STRING_FIXED, commentAsBytes);
            sendPacket.writeBytes(StringLengthDataType.STRING_FIXED, Constants.SPACE_STAR_SLASH_SPACE_AS_BYTES);
        }
        if (!this.platformDbCharsetMatches && StringUtils.startsWithIgnoreCaseAndWs(query, "LOAD DATA")) {
            sendPacket.writeBytes(StringLengthDataType.STRING_FIXED, StringUtils.getBytes(query));
        } else {
            sendPacket.writeBytes(StringLengthDataType.STRING_FIXED, StringUtils.getBytes(query, characterEncoding));
        }
        return sendQueryPacket(callingQuery, sendPacket, maxRows, streamResults, catalog, cachedMetadata, getProfilerEventHandlerInstanceFunction,
                resultSetFactory);
    }

public final <T extends Resultset> T sendQueryPacket(Query callingQuery, NativePacketPayload queryPacket, int maxRows, boolean streamResults,
            String catalog, ColumnDefinition cachedMetadata, GetProfilerEventHandlerInstanceFunction getProfilerEventHandlerInstanceFunction,
            ProtocolEntityFactory<T, NativePacketPayload> resultSetFactory) throws IOException {
        this.statementExecutionDepth++;

        byte[] queryBuf = null;
        int oldPacketPosition = 0;
        long queryStartTime = 0;
        long queryEndTime = 0;

        queryBuf = queryPacket.getByteBuffer();
        oldPacketPosition = queryPacket.getPosition(); // save the packet position

        queryStartTime = getCurrentTimeNanosOrMillis();

        LazyString query = new LazyString(queryBuf, 1, (oldPacketPosition - 1));

        try {

            if (this.queryInterceptors != null) {
                T interceptedResults = invokeQueryInterceptorsPre(query, callingQuery, false);

                if (interceptedResults != null) {
                    return interceptedResults;
                }
            }

            if (this.autoGenerateTestcaseScript) {
                StringBuilder debugBuf = new StringBuilder(query.length() + 32);
                generateQueryCommentBlock(debugBuf);
                debugBuf.append(query);
                debugBuf.append(';');
                TestUtils.dumpTestcaseQuery(debugBuf.toString());
            }

            // Send query command and sql query string
            NativePacketPayload resultPacket = sendCommand(queryPacket, false, 0);

            long fetchBeginTime = 0;
            long fetchEndTime = 0;

            String profileQueryToLog = null;

            boolean queryWasSlow = false;

            if (this.profileSQL || this.logSlowQueries) {
                queryEndTime = getCurrentTimeNanosOrMillis();

                boolean shouldExtractQuery = false;

                if (this.profileSQL) {
                    shouldExtractQuery = true;
                } else if (this.logSlowQueries) {
                    long queryTime = queryEndTime - queryStartTime;

                    boolean logSlow = false;

                    if (!this.useAutoSlowLog) {
                        logSlow = queryTime > this.propertySet.getIntegerProperty(PropertyKey.slowQueryThresholdMillis).getValue();
                    } else {
                        logSlow = this.metricsHolder.isAbonormallyLongQuery(queryTime);
                        this.metricsHolder.reportQueryTime(queryTime);
                    }

                    if (logSlow) {
                        shouldExtractQuery = true;
                        queryWasSlow = true;
                    }
                }

                if (shouldExtractQuery) {
                    // Extract the actual query from the network packet
                    boolean truncated = false;

                    int extractPosition = oldPacketPosition;

                    if (oldPacketPosition > this.maxQuerySizeToLog.getValue()) {
                        extractPosition = this.maxQuerySizeToLog.getValue() + 1;
                        truncated = true;
                    }

                    profileQueryToLog = StringUtils.toString(queryBuf, 1, (extractPosition - 1));

                    if (truncated) {
                        profileQueryToLog += Messages.getString("Protocol.2");
                    }
                }

                fetchBeginTime = queryEndTime;
            }

            T rs = readAllResults(maxRows, streamResults, resultPacket, false, cachedMetadata, resultSetFactory);

            long threadId = getServerSession().getCapabilities().getThreadId();
            int queryId = (callingQuery != null) ? callingQuery.getId() : 999;
            int resultSetId = rs.getResultId();
            long eventDuration = queryEndTime - queryStartTime;

            if (queryWasSlow && !this.serverSession.queryWasSlow() /* don't log slow queries twice */) {
                StringBuilder mesgBuf = new StringBuilder(48 + profileQueryToLog.length());

                mesgBuf.append(Messages.getString("Protocol.SlowQuery",
                        new Object[] { String.valueOf(this.useAutoSlowLog ? " 95% of all queries " : this.slowQueryThreshold), this.queryTimingUnits,
                                Long.valueOf(queryEndTime - queryStartTime) }));
                mesgBuf.append(profileQueryToLog);

                ProfilerEventHandler eventSink = getProfilerEventHandlerInstanceFunction.apply();

                eventSink.consumeEvent(
                        new ProfilerEventImpl(ProfilerEvent.TYPE_SLOW_QUERY, "", catalog, threadId, queryId, resultSetId, System.currentTimeMillis(),
                                eventDuration, this.queryTimingUnits, null, LogUtils.findCallingClassAndMethod(new Throwable()), mesgBuf.toString()));

                if (this.propertySet.getBooleanProperty(PropertyKey.explainSlowQueries).getValue()) {
                    if (oldPacketPosition < MAX_QUERY_SIZE_TO_EXPLAIN) {
                        queryPacket.setPosition(1); // skip first byte 
                        explainSlowQuery(query.toString(), profileQueryToLog);
                    } else {
                        this.log.logWarn(Messages.getString("Protocol.3", new Object[] { MAX_QUERY_SIZE_TO_EXPLAIN }));
                    }
                }
            }

            if (this.profileSQL || this.logSlowQueries) {

                ProfilerEventHandler eventSink = getProfilerEventHandlerInstanceFunction.apply();

                String eventCreationPoint = LogUtils.findCallingClassAndMethod(new Throwable());

                if (this.logSlowQueries) {
                    if (this.serverSession.noGoodIndexUsed()) {
                        eventSink.consumeEvent(
                                new ProfilerEventImpl(ProfilerEvent.TYPE_SLOW_QUERY, "", catalog, threadId, queryId, resultSetId, System.currentTimeMillis(),
                                        eventDuration, this.queryTimingUnits, null, eventCreationPoint, Messages.getString("Protocol.4") + profileQueryToLog));
                    }
                    if (this.serverSession.noIndexUsed()) {
                        eventSink.consumeEvent(
                                new ProfilerEventImpl(ProfilerEvent.TYPE_SLOW_QUERY, "", catalog, threadId, queryId, resultSetId, System.currentTimeMillis(),
                                        eventDuration, this.queryTimingUnits, null, eventCreationPoint, Messages.getString("Protocol.5") + profileQueryToLog));
                    }
                    if (this.serverSession.queryWasSlow()) {
                        eventSink.consumeEvent(new ProfilerEventImpl(ProfilerEvent.TYPE_SLOW_QUERY, "", catalog, threadId, queryId, resultSetId,
                                System.currentTimeMillis(), eventDuration, this.queryTimingUnits, null, eventCreationPoint,
                                Messages.getString("Protocol.ServerSlowQuery") + profileQueryToLog));
                    }
                }

                fetchEndTime = getCurrentTimeNanosOrMillis();

                eventSink.consumeEvent(new ProfilerEventImpl(ProfilerEvent.TYPE_QUERY, "", catalog, threadId, queryId, resultSetId, System.currentTimeMillis(),
                        eventDuration, this.queryTimingUnits, null, eventCreationPoint, profileQueryToLog));

                eventSink.consumeEvent(new ProfilerEventImpl(ProfilerEvent.TYPE_FETCH, "", catalog, threadId, queryId, resultSetId, System.currentTimeMillis(),
                        (fetchEndTime - fetchBeginTime), this.queryTimingUnits, null, eventCreationPoint, null));
            }

            if (this.hadWarnings) {
                scanForAndThrowDataTruncation();
            }

            if (this.queryInterceptors != null) {
                T interceptedResults = invokeQueryInterceptorsPost(query, callingQuery, rs, false);

                if (interceptedResults != null) {
                    rs = interceptedResults;
                }
            }

            return rs;
        } catch (CJException sqlEx) {
            if (this.queryInterceptors != null) {
                invokeQueryInterceptorsPost(query, callingQuery, null, false); // we don't do anything with the result set in this case
            }

            if (callingQuery != null) {
                callingQuery.checkCancelTimeout();
            }

            throw sqlEx;

        } finally {
            this.statementExecutionDepth--;
        }
    }