数据库类的简介

SQLiteOpenHelper：

管理SQLite的帮助类，提供获取SQLiteDatabase实例的方法，

它会在第一次使用数据库时调用获取实例方法时创建SQLiteDatabase实例，

并且处理数据库版本变化，开发人员在实现ContentProvider时都要实现一个自定义的SQLiteOpenHelper类，

处理数据的创建、升级和降级。

SQLiteDatabase：

代表一个打开的SQLite数据库，提供了执行数据库操作的接口方法。如果不需要在进程之间共享数据，

应用程序也可以自行创建这个类的实例来读写SQLite数据库。

SQLiteSession：

SQLiteSession负责管理数据库连接和事务的生命周期，

SQLiteSession通过SQLiteConnectionPool获取数据库连接,从而执行具体的数据库操作。

SQLiteConnectionPool：

数据库连接池，管理所有打开的数据库连接(Connection)。

所有数据库连接都是通过它来打开，打开后会加入连接池，

在读写数据库时需要从连接池中获取一个数据库连接来使用。

SQLiteConnection：

代表了数据库连接，每个Connection封装了一个native层的sqlite3实例，

通过JNI调用SQLite动态库的接口方法操作数据库，Connection要么被Session持有，要么被连接池持有。

CursorFactory：

可选的Cursor工厂，可以提供自定义工厂来创建Cursor。

DatabaseErrorHandler：

可选的数据库异常处理器(目前仅处理数据库Corruption)，如果不提供，将会使用默认的异常处理器。

SQLiteDatabaseConfiguration：

数据库配置，应用程序可以创建多个到SQLite数据库的连接，这个类用来保证每个连接的配置都是相同的。

SQLiteQuery和SQLiteStatement：

从抽象类SQLiteProgram派生，封装了SQL语句的执行过程，

在执行时自动组装待执行的SQL语句，并调用SQLiteSession来执行数据库操作。

打开数据库连接

通过 getReadableDatabase getWritableDatabase 方法 在使用的时候,

若未打开数据库,则创建数据库连接,

// SQLiteOpenHelper.java

public SQLiteDatabase getReadableDatabase() {

synchronized (this) {

//参数: writable

return getDatabaseLocked(false); //getWritableDatabase 传true

}

}

private SQLiteDatabase getDatabaseLocked(boolean writable) {

if (mDatabase != null) {

...

//check the invailed

return mDatabase;

}

... Isiniting

SQLiteDatabase db = mDatabase;

try {

try {

...

reopenReadWrite()

...

// transfer the path and params and open the db old code

db = SQLiteDatabase.openDatabase(filePath, params); //old code

// new code in wcdb

if (DEBUG_STRICT_READONLY && !writable) {

final String path = mContext.getDatabasePath(mName).getPath();

db = SQLiteDatabase.openDatabase(path, mPassword, mCipher, mFactory,

SQLiteDatabase.OPEN_READONLY, mErrorHandler, connectionPoolSize);

} else {

mNeedMode = true;

mMode = mEnableWriteAheadLogging ? Context.MODE_ENABLE_WRITE_AHEAD_LOGGING : 0;

db = Context.openOrCreateDatabase(mContext, mName, mPassword, mCipher,

mMode, mFactory, mErrorHandler, connectionPoolSize);

}

} catch (SQLException ex) {

// open db will try read-only

params = params.toBuilder().addOpenFlags(SQLiteDatabase.OPEN_READONLY).build();

// not wcdb

// db = SQLiteDatabase.openDatabase(filePath, params);

db = SQLiteDatabase.openDatabase(path, mPassword, mCipher, mFactory,

SQLiteDatabase.OPEN_READONLY, mErrorHandler);

}

}

...

onConfigure(db);

...

// about the version overwrite the method

onCreate

onUpgrade

}

//SQLiteConnectionPool.java

private SQLiteConnectionPool(SQLiteDatabase db, SQLiteDatabaseConfiguration configuration,

int poolSize) {

mDB = new WeakReference<>(db);

//数据库的配置信息

mConfiguration = new SQLiteDatabaseConfiguration(configuration);

//设置最大的数据库链接个数 //未开启wal的版本poolSize默认为1

setMaxConnectionPoolSizeLocked(poolSize);

}

流程：

SQLiteDatabase ->openOrCreateDatabase

-> openDatabase

->open->openInner

->SQLiteConnectionPool.open

->SQLiteConnectionPool.openConnectionLocked

->SQLiteConnection.open

private SQLiteConnection openConnectionLocked(SQLiteDatabaseConfiguration configuration,

boolean primaryConnection) {

//connectionId作为链接id，每次新创建一个数据库链接id自增1

final int connectionId = mNextConnectionId++;

//至此数据库链接被打开 SQLiteConnection.open

//Called by SQLiteConnectionPool only in method SQLiteConnectionPool#openConnectionLocked.

return SQLiteConnection.open(this, configuration,connectionId, primaryConnection); // might throw

}

创建数据库连接场景

创建或打开数据库，调用open

重新加载数据库配置，调用reconfigure(SQLiteDatabaseConfiguration configuration)

创建主链接，调用tryAcquirePrimaryConnectionLocked(int connectionFlags)

创建非主链接，调用tryAcquireNonPrimaryConnectionLocked(String sql, int connectionFlags)

SQLiteConnectionPool.png

数据库连接使用

//SQLiteDatabase.java

public long insert(String table, String nullColumnHack, ContentValues values) {

try {

//1.内部封装SQLiteStatement，2.调用statement.executeInsert();

return insertWithOnConflict(table, nullColumnHack, values, CONFLICT_NONE);

} catch (SQLException e) {

return -1;

}

}

//SQLiteStatement.java

public long executeInsert() {

acquireReference();

try {

//通过Session进行数据库操作

return getSession().executeForLastInsertedRowId(

getSql(), getBindArgs(), getConnectionFlags(), null);

} catch (SQLiteDatabaseCorruptException ex) {

checkCorruption(ex);

throw ex;

} finally {

releaseReference();

}

}

// Thread-local for database sessions that belong to this database.

// Each thread has its own database session.

// SQLiteSession的粒度是基于Thread的 每个线程都有一会话，且不可变。

// (Handler内部获取到当前线程的Looper)

private final ThreadLocal mThreadSession = new ThreadLocal() {

@Override

protected SQLiteSession initialValue() {

return createSession();

}

};

SQLiteSession getThreadSession() {

return mThreadSession.get(); // initialValue() throws if database closed

}

SQLiteSession createSession() {

final SQLiteConnectionPool pool;

synchronized (mLock) {

throwIfNotOpenLocked();

pool = mConnectionPoolLocked;

}

return new SQLiteSession(pool);

}

//SQLiteSession.java

/**

* Executes a statement that returns the row id of the last row inserted

* by the statement. Use for INSERT SQL statements.

* @param cancellationSignal A signal to cancel the operation in progress, or null if none.

* @return The row id of the last row that was inserted, or 0 if none.

*/

public long executeForLastInsertedRowId(String sql, Object[] bindArgs, int connectionFlags,

CancellationSignal cancellationSignal) {

//校验sql

if (sql == null) {

throw new IllegalArgumentException("sql must not be null.");

}

//对某些SQL语句(例如“ BEGIN”," COMMIT”和“ ROLLBACK”)执行特殊的重新解释,以确保事务状态不变式为保持。

if (executeSpecial(sql, bindArgs, connectionFlags, cancellationSignal)) {

return 0;

}

//获取数据库链接

acquireConnection(sql, connectionFlags, false, cancellationSignal); // might throw

try {

//使用链接执行数据库操作

return mConnection.executeForLastInsertedRowId(sql, bindArgs,

cancellationSignal); // might throw

} finally {

//释放数据库链接

releaseConnection(); // might throw

}

}

private void acquireConnection(String sql, int connectionFlags,

CancellationSignal cancellationSignal) {

if (mConnection == null) {

assert mConnectionUseCount == 0;

// 从连接池中获取数据库链接

mConnection = mConnectionPool.acquireConnection(sql, connectionFlags,

cancellationSignal); // might throw

mConnectionFlags = connectionFlags;

}

mConnectionUseCount += 1;

}

流程

SQLiteDatabase#insert

->SQLiteDatabase#insertWithOnConflict

->SQLiteStatement#executeInsert (构建sql语句)

->SQLiteSession#executeForLastInsertedRowId(通过Session进行数据库操作)

->SQLiteSession#acquireConnection

->SQLiteConnectionPool#acquireConnection

总结:

Connection从数据库连接池中获取的

进行数据库操作是通过Session操作Connection

多个线程执行数据库操作会有多个Session(通过ThreadLocal实现)

获取链接

//SQLiteConnectionPool.java

public SQLiteConnection acquireConnection(String sql, int connectionFlags,

CancellationSignal cancellationSignal) {

...

SQLiteConnection connection = waitForConnection(sql, connectionFlags, cancellationSignal);

...

SQLiteTrace callback = mTraceCallback;

if (callback != null) {

long waitTime = SystemClock.uptimeMillis() - startTime;

SQLiteDatabase db = mDB.get();

if (db != null) {

final boolean isPrimary =

(connectionFlags & CONNECTION_FLAG_PRIMARY_CONNECTION_AFFINITY) != 0;

callback.onConnectionObtained(db, sql, waitTime, isPrimary);

}

}

return connection;

}

private SQLiteConnection waitForConnection(String sql, int connectionFlags,

CancellationSignal cancellationSignal) {

//是否需要主链接

final boolean wantPrimaryConnection =

(connectionFlags & CONNECTION_FLAG_PRIMARY_CONNECTION_AFFINITY) != 0;

final ConnectionWaiter waiter;

final int nonce;

synchronized (mLock) {

throwIfClosedLocked();

// Abort if canceled.

//如果取消信号的回调不为空，那么执行回调检测是否需要取消

if (cancellationSignal != null) {

cancellationSignal.throwIfCanceled();

}

// Try to acquire a connection.

//尝试获得一个数据库链接

SQLiteConnection connection = null;

//不需要主链接，尝试获取非主链接

if (!wantPrimaryConnection) {

connection = tryAcquireNonPrimaryConnectionLocked(

sql, connectionFlags); // might throw

}

//获取不到非链接，尝试获取主链接

if (connection == null) {

connection = tryAcquirePrimaryConnectionLocked(connectionFlags); // might throw

}

if (connection != null) {

return connection;

}

// No connections available. Enqueue a waiter in priority order.

//没有可用的连接。按优先级进入等待队列排队

final int priority = getPriority(connectionFlags);

final long startTime = SystemClock.uptimeMillis();

//创建一个等待获取链接的对象

waiter = obtainConnectionWaiterLocked(Thread.currentThread(), startTime,

priority, wantPrimaryConnection, sql, connectionFlags);

ConnectionWaiter predecessor = null;

ConnectionWaiter successor = mConnectionWaiterQueue;

//按照优先级查找插入的位置

while (successor != null) {

if (priority > successor.mPriority) {

waiter.mNext = successor;

break;

}

predecessor = successor;

successor = successor.mNext;

}

//插入等待队列

if (predecessor != null) {

predecessor.mNext = waiter;

} else {

mConnectionWaiterQueue = waiter;

}

nonce = waiter.mNonce;

}

// Set up the cancellation listener.

//设置取消监听器，在等待的过程中如果取消等待那么执行cancelConnectionWaiterLocked

if (cancellationSignal != null) {

cancellationSignal.setOnCancelListener(new CancellationSignal.OnCancelListener() {

@Override

public void onCancel() {

synchronized (mLock) {

if (waiter.mNonce == nonce) {

//从等待队列中删除这个节点数据

//给waiter添加OperationCanceledException异常信息

//唤醒waiter对应线程的阻塞

//调用wakeConnectionWaitersLocked判断队列其他waiter是否状态有更新

cancelConnectionWaiterLocked(waiter);

}

}

}

});

}

try {

// Park the thread until a connection is assigned or the pool is closed.

// Rethrow an exception from the wait, if we got one.

//驻留线程，直到分配了连接或关闭了池。

//如果有异常，则从等待中抛出异常。

long busyTimeoutMillis = CONNECTION_POOL_BUSY_MILLIS;

long nextBusyTimeoutTime = waiter.mStartTime + busyTimeoutMillis;

for (;;) {

// Detect and recover from connection leaks.

//是否需要从泄漏中进行恢复，之前被调用onConnectionLeaked //mConnectionLeaked.set(true);

// 获取连接的策略: 定期检查此标志，以便它可以从泄漏的连接中恢复并唤醒

if (mConnectionLeaked.compareAndSet(true, false)) {

synchronized (mLock) {

//为等待数据库链接队列进行链接赋值

wakeConnectionWaitersLocked();

}

}

// how to stop?

// Wait to be unparked (may already have happened), a timeout, or interruption.

//阻塞busyTimeoutMillis毫秒，或者中间被执行LockSupport.unpark (释放许可)

//被执行cancelConnectionWaiterLocked进行取消

//或者被执行wakeConnectionWaitersLocked进行链接分配

LockSupport.parkNanos(this, busyTimeoutMillis * 1000000L);

// Clear the interrupted flag, just in case.

Thread.interrupted();//重置当前线程的中断状态

// Check whether we are done waiting yet.

//检查我们是否已经完成等待。

synchronized (mLock) {

throwIfClosedLocked();//如果数据库关闭，那么抛出异常

final SQLiteConnection connection = waiter.mAssignedConnection;

final RuntimeException ex = waiter.mException;

//如果已经分配到链接，或者异常

if (connection != null || ex != null) {

recycleConnectionWaiterLocked(waiter);//回收waiter

if (connection != null) {//返回分配链接

return connection;

}

throw ex; // rethrow!重新抛出异常

}

final long now = SystemClock.uptimeMillis();

if (now < nextBusyTimeoutTime) {

//parkNanos阻塞时间不够busyTimeoutMillis毫秒，被执行LockSupport.unpark

busyTimeoutMillis = now - nextBusyTimeoutTime;

} else {

busyInfo = gatherConnectionPoolBusyInfoLocked();

//重置下次阻塞时间

busyTimeoutMillis = CONNECTION_POOL_BUSY_MILLIS;

nextBusyTimeoutTime = now + busyTimeoutMillis;

}

if (busyInfo != null) {

long waitMillis = now - waiter.mStartTime;

//输出日志

logConnectionPoolBusy(busyInfo, waitMillis, connectionFlags);

SQLiteDatabase db = mDB.get();

SQLiteTrace callback = mTraceCallback;

if (db != null && callback != null) {

// SQLiteTrace 性能监控的回调

callback.onConnectionPoolBusy(db, sql, waitMillis, wantPrimaryConnection,

busyInfo.activeSql, busyInfo.activeTransactions);

}

}

}

}

} finally {

// Remove the cancellation listener.

//有异常，或者获取到分配的链接 那么解绑"取消信号回调"信息

if (cancellationSignal != null) {

cancellationSignal.setOnCancelListener(null);

}

}

}

acquireDBConnection.png

LockSupport.parkNanos 循环判断是否获得了数据库链接,未获取到则继续睡眠,

直到这次操作被取消或者获得数据库链接。

LockSupport.park() 的实现原理是通过二元信号量实现阻塞，

这个信号量最多只能加到1。可以理解为获取释放许可证的场景。

unpark()方法会释放一个许可证，park()方法则是获取许可证，如果当前没有许可证，

则进入休眠状态，知道许可证被释放了才被唤醒。无论执行多少次unpark()方法，也最多只会有一个许可证。

主链接的获取

//SQLiteConnectionPool.java

@GuardedBy("mLock")

private SQLiteConnection tryAcquirePrimaryConnectionLocked(int connectionFlags) {

// If the primary connection is available, acquire it now.

SQLiteConnection connection = mAvailablePrimaryConnection;

if (connection != null) {

mAvailablePrimaryConnection = null;

finishAcquireConnectionLocked(connection, connectionFlags); // might throw

return connection;

}

// Make sure that the primary connection actually exists and has just been acquired.

for (SQLiteConnection acquiredConnection : mAcquiredConnections.keySet()) {

if (acquiredConnection.isPrimaryConnection()) {

return null;

}

}

// Uhoh. No primary connection! Either this is the first time we asked

// for it, or maybe it leaked?

//第一次创建数据库主链接，或者主链接被回收

connection = openConnectionLocked(mConfiguration,

true /*primaryConnection*/); // might throw

finishAcquireConnectionLocked(connection, connectionFlags); // might throw

return connection;

}

总结: 主数据库链接只一个，如果被占用那么需要等待，如果没有那么就需要创建。

获取非主链接

//SQLiteConnectionPool.java

private SQLiteConnection tryAcquireNonPrimaryConnectionLocked(

String sql, int connectionFlags) {

// Try to acquire the next connection in the queue.

SQLiteConnection connection;

//尝试获取队列中的下一个连接。

final int availableCount = mAvailableNonPrimaryConnections.size();

if (availableCount > 1 && sql != null) {

// If we have a choice, then prefer a connection that has the

// prepared statement in its cache.

//检查是否可以在其缓存中选择包含prepare语句的连接。

for (int i = 0; i < availableCount; i++) {

connection = mAvailableNonPrimaryConnections.get(i);

if (connection.isPreparedStatementInCache(sql)) {

// 从非主链接集合中移除

mAvailableNonPrimaryConnections.remove(i);

finishAcquireConnectionLocked(connection, connectionFlags); // might throw

return connection;

}

}

}

if (availableCount > 0) {

// Otherwise, just grab the last one.//next one 抓取非主链接集合中的最后一个

connection = mAvailableNonPrimaryConnections.remove(availableCount - 1);

finishAcquireConnectionLocked(connection, connectionFlags); // might throw

return connection;

}

//如果没有可以的非主链接，则扩展数据库连接池

// Expand the pool if needed.

int openConnections = mAcquiredConnections.size();

if (mAvailablePrimaryConnection != null) {

openConnections += 1;

}

//如果数据库连接池已经达到上限，返回null

if (openConnections >= mMaxConnectionPoolSize) {

return null;

}

//否则创建新的非主链接

connection = openConnectionLocked(mConfiguration,

false /*primaryConnection*/); // might throw

finishAcquireConnectionLocked(connection, connectionFlags); // might throw

return connection;

}

非主数据库链接数量的多少受限于数据库连接池的大小。

数据库链接释放

//SQLiteConnectionPool.java

// Weak references to all acquired connections. The associated value

// indicates whether the connection must be reconfigured before being

// returned to the available connection list or discarded.

// For example, the prepared statement cache size may have changed and

// need to be updated in preparation for the next client.

private final WeakHashMap mAcquiredConnections

= new WeakHashMap<>();

//释放数据库链接,使其返回连接池

public void releaseConnection(SQLiteConnection connection) {

synchronized (mLock) {

//idle事件处理connectionReleased

if (mIdleConnectionHandler != null) {

mIdleConnectionHandler.connectionReleased(connection);

}

//获取这个链接的状态 by invoke recycleConnectionLocked

//NORMAL,正常返回连接池

//RECONFIGURE,必须先重新配置连接，然后才能返回。

//DISCARD,连接必须关闭并丢弃。

AcquiredConnectionStatus status = mAcquiredConnections.remove(connection);

if (status == null) {

throw new IllegalStateException("Cannot perform this operation "

+ "because the specified connection was not acquired "

+ "from this pool or has already been released.");

}

//检测是否已经关闭连接池

if (!mIsOpen) {

closeConnectionAndLogExceptionsLocked(connection);

} else if (connection.isPrimaryConnection()) {//如果是主链接

//判断这个数据库链接是否需要回收

if (recycleConnectionLocked(connection, status)) {

assert mAvailablePrimaryConnection == null;

mAvailablePrimaryConnection = connection;//标识主链接可用，被占用的时候为null

}

//判断队列其他waiter是否状态有更新

wakeConnectionWaitersLocked();

} else if (mAvailableNonPrimaryConnections.size() >= mMaxConnectionPoolSize - 1) {

//可用的非主链接数+主链接>=最大链接数的时 关闭这个链接

closeConnectionAndLogExceptionsLocked(connection);

} else {

//判断这个数据库链接是否需要回收 不回收则添加

if (recycleConnectionLocked(connection, status)) {

//将这个链接添加到非主链接容器中

mAvailableNonPrimaryConnections.add(connection);

}

//判断队列其他waiter是否状态有更新

wakeConnectionWaitersLocked();

}

}

}

//唤醒waiter对应的线程 取消已获取许可的park

//将异常赋值给waiter进行抛出

@GuardedBy("mLock")

private void wakeConnectionWaitersLocked() {

// 释放等待

// Unpark all waiters that have requests that we can fulfill.

// This method is designed to not throw runtime exceptions, although we might send

// a waiter an exception for it to rethrow.

ConnectionWaiter predecessor = null;

//链表的head point

ConnectionWaiter waiter = mConnectionWaiterQueue;

boolean primaryConnectionNotAvailable = false;

boolean nonPrimaryConnectionNotAvailable = false;

while (waiter != null) {

boolean unpark = false;

//是否关闭了数据库，如果关闭了那么唤醒所有waiter的线程

if (!mIsOpen) {

unpark = true;

} else {

try {

SQLiteConnection connection = null;

//如果该waiter需要非主链接，而且现在有可用的非主链接

if (!waiter.mWantPrimaryConnection && !nonPrimaryConnectionNotAvailable) {

//获取非主链接

connection = tryAcquireNonPrimaryConnectionLocked(

waiter.mSql, waiter.mConnectionFlags); // might throw

//获取为空，标识现在没有可用的非主链接

if (connection == null) {

nonPrimaryConnectionNotAvailable = true;

}

}

//主链接可以用

if (connection == null && !primaryConnectionNotAvailable) {

//尝试获取主链接

connection = tryAcquirePrimaryConnectionLocked(

waiter.mConnectionFlags); // might throw

//获取为空，标识现在主链接不可用

if (connection == null) {

primaryConnectionNotAvailable = true;

}

}

//获取到了数据库链接

if (connection != null) {

waiter.mAssignedConnection = connection;//改waiter赋值链接

unpark = true;//唤醒该waiter的对应线程

} else if (nonPrimaryConnectionNotAvailable && primaryConnectionNotAvailable) {

// There are no connections available and the pool is still open.

// We cannot fulfill any more connection requests, so stop here.

//连接池仍然可用，但是没有可用的链接,无法对其他的waiter状态进行更新则直接返回

break;

}

} catch (RuntimeException ex) {

// Let the waiter handle the exception from acquiring a connection.

waiter.mException = ex;

unpark = true;

}

}

final ConnectionWaiter successor = waiter.mNext;

//如果需要唤醒，那么从链表中删除这个waiter，并进行对应线程唤醒操作

if (unpark) {

if (predecessor != null) {

predecessor.mNext = successor;

} else {

mConnectionWaiterQueue = successor;

}

waiter.mNext = null;

LockSupport.unpark(waiter.mThread);

} else {

predecessor = waiter;

}

waiter = successor;

}

}

链接的释放有时候是为了回收，有时候为了重用。重用的时候还需要唤醒等待链接队列中获得这个链接的waiter 。

数据库链接池的关闭

数据库的关闭

//SQLiteClosable.java，它是SQLiteDatabase的父类

//释放引用的对象，直到所有的引用都被释放了那么关闭数据库

public void close() {

releaseReference();

}

public void releaseReference() {

boolean refCountIsZero;

synchronized (this) {

refCountIsZero = --mReferenceCount == 0;

}

//引用计数递减

if (refCountIsZero) {

onAllReferencesReleased();

}

}

//SQLiteDatabase.java

@Override

protected void onAllReferencesReleased() {

dispose(false);

}

private void dispose(boolean finalized) {

final SQLiteConnectionPool pool;

synchronized (mLock) {

pool = mConnectionPoolLocked;

//连接池置空，无法进行新操作

mConnectionPoolLocked = null;

}

if (!finalized) {

synchronized (sActiveDatabases) {

//删除当前数据库的引用

sActiveDatabases.remove(this);

}

// 触发连接池的关闭

if (pool != null) {

pool.close(); //SQLiteConnectionPool.close()

}

}

}

数据库连接池的关闭是由数据库关闭触发的。

数据库链接池的关闭

//关闭数据库连接池，停止接受新的数据库链接的请求。

//链接池中的可用链接立即被关闭，其他正在使用的链接被归还到数据的时候关闭

public void close() {

dispose(false);

}

private void dispose(boolean finalized) {

if (!finalized) {

// Close all connections. We don't need (or want) to do this

// when finalized because we don't know what state the connections

// themselves will be in. The finalizer is really just here for CloseGuard.

// The connections will take care of themselves when their own finalizers run.

synchronized (mLock) {

throwIfClosedLocked();//检测是否已经被关闭

mIsOpen = false;//标识数据库连接池关闭

//关闭数据库连接池中目前可用的链接(可用的主链接与非主链接集合)

closeAvailableConnectionsAndLogExceptionsLocked(); //SQLiteConnection.close()

final int pendingCount = mAcquiredConnections.size();

//仍然有链接正在使用中

if (pendingCount != 0) {

Log.i(TAG, "The connection pool for " + mConfiguration.label

+ " has been closed but there are still "

+ pendingCount + " connections in use. They will be closed "

+ "as they are released back to the pool.");

}

//队列其他waiter是否状态有更新

wakeConnectionWaitersLocked();

}

}

}

数据库链接的关闭

//SQLiteConnectionPool.java

// Can't throw.

@GuardedBy("mLock")

private void closeAvailableConnectionsAndLogExceptionsLocked() {

//关闭可用的非主链接

closeAvailableNonPrimaryConnectionsAndLogExceptionsLocked();

//关闭主链接

if (mAvailablePrimaryConnection != null) {

closeConnectionAndLogExceptionsLocked(mAvailablePrimaryConnection);

mAvailablePrimaryConnection = null;

}

}

// Can't throw.

@GuardedBy("mLock")

private void closeAvailableNonPrimaryConnectionsAndLogExceptionsLocked() {

//循环遍历可用的非主链接，关闭链接

final int count = mAvailableNonPrimaryConnections.size();

for (int i = 0; i < count; i++) {

closeConnectionAndLogExceptionsLocked(mAvailableNonPrimaryConnections.get(i));

}

mAvailableNonPrimaryConnections.clear();

}

// Can't throw.

@GuardedBy("mLock")

private void closeConnectionAndLogExceptionsLocked(SQLiteConnection connection) {

try {

connection.close(); // might throw

if (mIdleConnectionHandler != null) {

mIdleConnectionHandler.connectionClosed(connection);

}

} catch (RuntimeException ex) {

Log.e(TAG, "Failed to close connection, its fate is now in the hands "

+ "of the merciful GC: " + connection, ex);

}

}

数据库关闭的时候引用次数自减，若引用次数归零则真正执行关闭数据库；

数据库关闭清除引用后,进行数据库连接池的关闭；

关闭所有的空闲链接，使用中的连接回归连接池后被关闭；

并发问题的解决

历史实现： SQLite的原子操作实现

默认方法是 rollback journal。

当对 DB 进行写操作的时候，SQLite 首先将准备要修改部分的原始内容(以 Page 为单位)拷贝到“回滚日志”中，用于后续可能的 Rollback操作以及 Crash、断电等意外造成写操作中断时恢复 DB 的原始状态，回滚日志存放于名为“DB文件名-journal”的独立文件中(以下简称“-journal”)。对原始内容做备份后，才能写入修改后的内容到 DB 主文件中，

当写入操作完成，用户提交事务后，SQLite 清空 -journal 的内容，至此完成一个完整的写事务。

rollback_journal.png

Rollback 模式中，每次拷贝原始内容或写入新内容后，都需要确保之前写入的数据真正写入到磁盘，而不是缓存在操作系统中，这需要发起一次 fsync 操作，通知并等待操作系统将缓存真正写入磁盘，这个过程十分耗时。

除了耗时的 fsync 操作，写入 -journal 以及 DB 主文件的时候，是需要独占整个 DB 的，否则别的线程/进程可能读取到写到一半的内容。这样的设计使得写操作与读操作是互斥的，并发性很差。

WAL模式和异步Checkpoint

WCDB 开启了 WAL 模式

针对WAL做了改进,使用异步Checkpoint

WAL 模式+则改变了上述流程，写操作不直接写入 DB 主文件，而是写到“DB文件名-wal”文件的末尾，并且通过 -shm 共享内存文件来实现 -wal 内容的索引。读操作时，将结合 DB 主文件以及 -wal 的内容返回结果。由于读操作只读取 DB 主文件和 -wal 前面非正在写入的部分，不需要读取写操作正在写到一半的内容，WAL 模式下读与写操作的并发由此实现。

WAL 写操作除了上面的流程，还增加了一步：Checkpoint，即将 -wal 的内容与合并到 DB 主文件。 由于写操作将内容临时写到 -wal 文件，-wal 文件会不断增大且拖慢读操作，因此需要定期进行 Checkpoint 操作将 -wal 文件保持在合理的大小。Checkpoint 操作比较耗时且会阻塞读操作，但由于时效性要求较低，遇到堵塞可以暂时放弃Checkpoint 操作. 继续 DB 读写操作，不至于太过影响读写性能。SQLite 官方默认的 Checkpoint 阈值是 1000 page，即当 -wal 文件达到 1000 page 大小时，写操作的线程在完成写操作后,再同步进行 Checkpoint 操作；Android Framework 的 Checkpoint 阈值是 100 page。

wal.png

基于 WAL 的基本工作方式，两个优化点：

写入 -wal 文件时不进行 fsync 操作，因为 -wal 文件损坏只影响新写入的未 Checkpoint 部分数据而非整个数据库损坏，影响相对小

将需要进行fsync的Checkpoint 操作放到独立线程执行，让写操作能尽快返回

异步Checkpoint的基本思路，减少和转移耗时较多而且性能不稳定的 fsync 操作，增强写操作性能和减少突然卡顿的可能性，同时不增加 DB 损坏率。

// SQLiteConnectionPool.java

SQLiteCheckpointListener getCheckpointListener() {

return mCheckpointListener;

}

void setCheckpointListener(SQLiteCheckpointListener listener) {

SQLiteDatabase db = mDB.get();

if (mCheckpointListener != null)

mCheckpointListener.onDetach(db);

mCheckpointListener = listener;

if (mCheckpointListener != null)

mCheckpointListener.onAttach(db);

}

// 调用通过jni

/*package*/ void notifyCheckpoint(String dbName, int pages) {

SQLiteDatabase db = mDB.get();

SQLiteCheckpointListener walHook = mCheckpointListener;

if (walHook == null || db == null)

return;

walHook.onWALCommit(db, dbName, pages);

}

public interface SQLiteCheckpointListener {

void onAttach(SQLiteDatabase db);

//Called immediately when a WAL transaction has been committed

void onWALCommit(SQLiteDatabase db, String dbName, int pages);

void onDetach(SQLiteDatabase db);

}

// 实现类: SQLiteAsyncCheckpointer

public SQLiteAsyncCheckpointer(Looper looper, int threshold, int blockingThreshold) {

mLooper = looper;

mThreshold = threshold;

mBlockingThreshold = blockingThreshold;

mPendingCheckpoints = new HashSet<>();

}

@Override

public void onAttach(SQLiteDatabase db) {

if (mLooper == null) {

mLooper = acquireDefaultLooper();

mUseDefaultLooper = true;

}

mHandler = new Handler(mLooper, this);

mLastSyncMode = db.getSynchronousMode();

db.setSynchronousMode(SQLiteDatabase.SYNCHRONOUS_NORMAL);

}

/**

* shmily add note

* 此处要关注WAL的提交策略

*/

@Override

public void onWALCommit(SQLiteDatabase db, String dbName, int pages) {

if (pages < mThreshold)

return;

boolean blockWriting = pages >= mBlockingThreshold;

Pair p = new Pair<>(db, dbName);

boolean newTask;

synchronized (mPendingCheckpoints) {

newTask = mPendingCheckpoints.add(p);

}

if (!newTask)

return;

/**

* shmily add note

* acquireReference 与 releaseReference 成对调用

* 用于引用计数 引用则次数+1 释放则次数-1

* 在sendMessage之前acquireReference

* handMessage处理完成后释放

*/

db.acquireReference();

Message msg = mHandler.obtainMessage(0, blockWriting ? 1 : 0, 0, p);

mHandler.sendMessage(msg);

}

@Override

public void onDetach(SQLiteDatabase db) {

// 恢复SynchronousMode

db.setSynchronousMode(mLastSyncMode);

mHandler = null;

if (mUseDefaultLooper) {

mLooper = null;

releaseDefaultLooper();

mUseDefaultLooper = false;

}

}

@Override

@SuppressWarnings("unchecked")

public boolean handleMessage(Message msg) {

Pair p = (Pair) msg.obj;

// Pair 的方便之处

SQLiteDatabase db = p.first;

String dbName = p.second;

boolean blockWriting = msg.arg1 != 0;

try {

long time = SystemClock.uptimeMillis();

Pair result = db.walCheckpoint(dbName, blockWriting);

int walPages = result.first;

int checkpointedPages = result.second;

time = SystemClock.uptimeMillis() - time;

onCheckpointResult(db, walPages, checkpointedPages, time);

} finally {

db.releaseReference();

}

synchronized (mPendingCheckpoints) {

if (!mPendingCheckpoints.remove(p))

throw new AssertionError("mPendingCheckpoints.remove(p)");

}

return true;

}

protected void onCheckpointResult(SQLiteDatabase db, int walPages, int checkpointedPages,

long time) {

// implemented by the derived class

}

private static Looper acquireDefaultLooper() {

synchronized (gDefaultThreadLock) {

if (gDefaultThreadRefCount++ == 0) {

// Initialize default handler thread.

if (gDefaultThread != null)

throw new AssertionError("gDefaultThread == null");

gDefaultThread = new HandlerThread("WCDB.AsyncCheckpointer", 4);

gDefaultThread.start();

}

return gDefaultThread.getLooper();

}

}

private static void releaseDefaultLooper() {

synchronized (gDefaultThreadLock) {

if (--gDefaultThreadRefCount <= 0) {

if (gDefaultThreadRefCount < 0)

throw new AssertionError("gDefaultThreadRefCount == 0");

gDefaultThread.quit();

gDefaultThread = null;

}

}

}