Android数据库源码分析(2)-SQLiteDatabase的实现以及多线程行为

Android数据库源码分析(2)-SQLiteDatabase的实现以及多线程行为

 

本系列主要关注安卓数据库的线程行为，分为四个部分：
(1)SQLiteOpenHelper的getReadableDatabase和getWritableDatabase
(2)SQLiteDatabase的实现以及多线程行为
(3)连接缓存池SQLiteConnectionPool
(4)SQLiteDatabase多线程实践

本篇主要关注SQLiteDatabase的线程同步实现与架构实现。

1 SQLiteClosable的acquireReference与releaseReference方法

SQLiteClosable是SQLiteDatabase的父类，也同时是数据库下其他几个类的父类。其中实现了引用计数逻辑来控制资源释放的时机。

private int mReferenceCount = 1;

public void acquireReference() {
    synchronized(this) {
        if (mReferenceCount <= 0) {
            throw new IllegalStateException(
                    "attempt to re-open an already-closed object: " + this);
        }
        mReferenceCount++;
    }
}

public void releaseReference() {
    boolean refCountIsZero = false;
    synchronized(this) {
        refCountIsZero = --mReferenceCount == 0;
    }
    if (refCountIsZero) {
        onAllReferencesReleased();
    }
}

可以看到这里用mReferenceCount简单地实现了一个引用计数。而引用计数的初始值是1。SQLiteDatabase会在每次操作前调用一次acquireReference，而在结束后调用一次releaseReference。为了方便，下文中把这样的被acquireReference和releaseReference包裹的过程称为一次“操作”。
那么如果这两个方法保持成对调用的话，是不是就不可能触发onAllReferenceReleased方法？事实上，SQLiteClosable还有一个方法close调用了releaseReference。由于锁的存在，只要不在其它“操作”中调用close，调用close之后mReferenceCount的值可以断定是0。
到这里为止，感觉上是可以用一个boolean值来标记引用状态的。因为由于锁的存在，只要各个“操作”是序列进行的（没有一个“操作”调用了另一个“操作”的情况），mReferenceCount只可能是0和1。推测引用计数就是为了应付“操作”之间存在调用这种情况。这就像同一个线程里的嵌套锁需要进行计数一样。

2 SQLiteDatabase的打开与关闭

2.1 关闭

上文中提到的onAllReferenceReleased是一个抽象方法。其在SQLiteDatabase中的实现为

@Override
protected void onAllReferencesReleased() {
    dispose(false);
}

在finalize中同样调用了dispose方法

protected void finalize() throws Throwable {
    try {
        dispose(true);
    } finally {
        super.finalize();
    }
}

而dispose的实现为

private void dispose(boolean finalized) {
    final SQLiteConnectionPool pool;
    synchronized (mLock) {
        if (mCloseGuardLocked != null) {
            if (finalized) {
                //CloseGuard是一个监测是否及时调用close方法的类，一般来说除了输出日志并不会做别的什么
                //这里事实上就是在finalize的时候如果没有close过，就输出一条日志
                mCloseGuardLocked.warnIfOpen();
            }
            mCloseGuardLocked.close();
        }
        pool = mConnectionPoolLocked;//这个mConnectionPool是连接池。此方法里将其置空并关闭。后文详细讨论其作用。
        mConnectionPoolLocked = null;
    }
    if (!finalized) {
        //sActiveDatabases是一个静态的WeakHashMap，用key来放置所有活动数据库，而value并没有作用。dispose的时候自然要移除this。
        //跟踪代码分析下来，用这个map只是为了bug report
        synchronized (sActiveDatabases) {
            sActiveDatabases.remove(this);
        }
        if (pool != null) {
            pool.close();
        }
    }
}

2.2 打开

在本系列第一篇中我们曾看到过，最终的打开数据库的是一个静态方法，SQLiteDatabase.openDatabase(String path, CursorFactory factory, int flags, DatabaseErrorHandler errorHandler)。

public static SQLiteDatabase openDatabase(String path, CursorFactory factory, int flags,
        DatabaseErrorHandler errorHandler) {
    SQLiteDatabase db = new SQLiteDatabase(path, flags, factory, errorHandler);
    db.open();
    return db;
}

这里很简单，就是新建一个对象，然后调用open。构造器里只有一些初始化，略过。着重看open方法：

private void open() {
    try {
        try {
            openInner();//尝试一次
        } catch (SQLiteDatabaseCorruptException ex) {
            onCorruption();//失败了，再次尝试前调用另一个方法。
            openInner();
        }
    } catch (SQLiteException ex) {
        Log.e(TAG, "Failed to open database '" + getLabel() + "'.", ex);
        close();
        throw ex;
    }
}

private void openInner() {
    synchronized (mLock) {
        assert mConnectionPoolLocked == null;
        mConnectionPoolLocked = SQLiteConnectionPool.open(mConfigurationLocked);
        mCloseGuardLocked.open("close");
    }

    synchronized (sActiveDatabases) {//这是之前那个WeakHashMap
        sActiveDatabases.put(this, null);
    }
}

void onCorruption() {
    EventLog.writeEvent(EVENT_DB_CORRUPT, getLabel());
    mErrorHandler.onCorruption(this);
}

open中会尝试调用openInner。如果失败一次，则调用onCorruption，随后再尝试一次。mErrorHandler是构造器传入的，构造器参数由静态方法openDatabase传入，而这个参数又最终从SQLiteOpenHelper传入。

openInner中做的事情，从命名上看，是开启一个SQLiteConnectionPool即数据库连接池。简单地说，数据库连接池维持了对数据库的多个连接。数据库连接的类是SQLiteConnection。

3 线程内单例的SQLiteSession

private final ThreadLocal<SQLiteSession> mThreadSession = new ThreadLocal<SQLiteSession>() {
    @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);
}

ThreadLocal会在每个线程内维护一个对象，而在线程结束时解除对对象的引用。initialValue方法会在线程中不存在已有对象时创建一个，不Override的话会给出一个null。除此之外也可以通过ThreadLocal.set来给本线程配置一个对象。
可以看到mThreadSession是一个ThreadLocal。调用getThreadSession会获取一个线程内单例的SQLiteSession对象。

SQLiteSession是提供数据库操作能力（增删改查以及事务）的一个单元。它会从SQLiteConnectionPool即连接池中获取连接，最终对数据库进行操作。

到这儿类已经有点多了。整理一下逻辑：
（1）SQLiteDatabase持有一个ThreadLocal，用于对每个线程生成一个SQLiteSession；
（2）SQLiteSession持有SQLiteConnectionPool(虽然SQLiteDatabase也持有连接池对象，但它只用来传递给SQLiteSession)，但是同一个SQLiteDatabase下的SQLiteSession是共用一个SQLiteConnectionPool的；
（3）SQLiteConnectionPool管理SQLiteConnection并适时向SQLiteSession提供之；
（4）SQLiteConnection直接对底层数据库进行操作(这个类里面才有大量的native方法)。

接下来分析一下SQLiteSession。

获取与释放连接，还是一个引用计数实现：

private final SQLiteConnectionPool mConnectionPool;//构造器中初始化，值从SQLiteDatabase对象中传入
private SQLiteConnection mConnection;
private int mConnectionFlags;
private int mConnectionUseCount;//无处不在的引用计数

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;
}

private void releaseConnection() {
    assert mConnection != null;
    assert mConnectionUseCount > 0;
    if (--mConnectionUseCount == 0) {
        try {
            mConnectionPool.releaseConnection(mConnection); // might throw
        } finally {
            mConnection = null;
        }
    }
}

具体的数据库操作有很多executeXXX形式的方法，逻辑大同小异。挑一个看看：

public int executeForChangedRowCount(String sql, Object[] bindArgs, int connectionFlags,
        CancellationSignal cancellationSignal) {
    if (sql == null) {
        throw new IllegalArgumentException("sql must not be null.");
    }

    if (executeSpecial(sql, bindArgs, connectionFlags, cancellationSignal)) {//排除特殊操作
        return 0;
    }

    //获取连接
    acquireConnection(sql, connectionFlags, cancellationSignal); // might throw
    try {
        //底层数据库操作。本文不关心。
        return mConnection.executeForChangedRowCount(sql, bindArgs,
                cancellationSignal); // might throw
    } finally {
        //释放连接
        releaseConnection(); // might throw
    }
}

//用来支持'BEGIN'，'COMMIT'，'ROLLBACK'的操作。就是与Transaction相关的操作。
private boolean executeSpecial(String sql, Object[] bindArgs, int connectionFlags,
        CancellationSignal cancellationSignal) {
    if (cancellationSignal != null) {
        cancellationSignal.throwIfCanceled();
    }

    final int type = DatabaseUtils.getSqlStatementType(sql);
    switch (type) {
        case DatabaseUtils.STATEMENT_BEGIN:
            beginTransaction(TRANSACTION_MODE_EXCLUSIVE, null, connectionFlags,
                    cancellationSignal);
            return true;

        case DatabaseUtils.STATEMENT_COMMIT:
            setTransactionSuccessful();
            endTransaction(cancellationSignal);
            return true;

        case DatabaseUtils.STATEMENT_ABORT:
            endTransaction(cancellationSignal);
            return true;
    }
    return false;
}

4 单次完整的SQLite操作

4.1 SQLiteStatement

以最简单的delete方法为例。其它方法的流程均大同小异。

public int delete(String table, String whereClause, String[] whereArgs) {
    acquireReference();
    try {
        SQLiteStatement statement =  new SQLiteStatement(this, "DELETE FROM " + table +
                (!TextUtils.isEmpty(whereClause) ? " WHERE " + whereClause : ""), whereArgs);
        try {
            return statement.executeUpdateDelete();
        } finally {
            statement.close();
        }
    } finally {
        releaseReference();
    }
}

先用SQLiteStatement做一些sql转义和拼接，然后调用statement.executeUpdateDelete()。

具体看一下executeUpdateDelete：

//以下来自SQLiteStatement
public int executeUpdateDelete() {
    acquireReference();//注意这里是SQLiteStatement内的引用计数，不是SQLiteDatabase了。
    try {
        return getSession().executeForChangedRowCount(
                getSql(), getBindArgs(), getConnectionFlags(), null);//上一节分析过了，执行SQL。
    } catch (SQLiteDatabaseCorruptException ex) {
        onCorruption();
        throw ex;
    } finally {
        releaseReference();
    }
}

//这个方法在父类SQLiteProgram中。又回到了上一小节的getThreadSession。获取线程内的单例。
protected final SQLiteSession getSession() {
    return mDatabase.getThreadSession();
}

4.2 SQLiteDirectCursorDriver与SQLiteQuery

与4.1不同的是，在进行query操作时，最终没有使用SQLiteStatement类，而是通过SQLiteDirectCursorDriver间接使用了SQLiteQuery。而SQLiteQuery与SQLiteStatement同为SQLiteProgram的子类，完成类似的功能。

所有的query操作最终均调用这样一个方法：

public Cursor rawQueryWithFactory(
        CursorFactory cursorFactory, String sql, String[] selectionArgs,
        String editTable, CancellationSignal cancellationSignal) {
    acquireReference();
    try {
        SQLiteCursorDriver driver = new SQLiteDirectCursorDriver(this, sql, editTable,
                cancellationSignal);
        return driver.query(cursorFactory != null ? cursorFactory : mCursorFactory,
                selectionArgs);
    } finally {
        releaseReference();
    }
}

而SQLiteDirectCursorDriver的query方法如下：

public Cursor query(CursorFactory factory, String[] selectionArgs) {
    final SQLiteQuery query = new SQLiteQuery(mDatabase, mSql, mCancellationSignal);
    final Cursor cursor;
    try {
        query.bindAllArgsAsStrings(selectionArgs);

        if (factory == null) {
            cursor = new SQLiteCursor(this, mEditTable, query);
        } else {
            cursor = factory.newCursor(mDatabase, this, mEditTable, query);
        }
    } catch (RuntimeException ex) {
        query.close();
        throw ex;
    }

    mQuery = query;
    return cursor;
}

其中新建了一个SQLiteQuery，并绑定参数。随后新建一个Cursor，这就是最终返回的Cursor对象。接下来考察无CursorFactory情况下默认返回的SQLiteCursor。

AbstractCursor中各种move方法均会调用moveToPosition，而moveToPosition会调用onMove，SQliteCursor中onMove的实现为：

@Override
public boolean onMove(int oldPosition, int newPosition) {
    // Make sure the row at newPosition is present in the window
    if (mWindow == null || newPosition < mWindow.getStartPosition() ||
            newPosition >= (mWindow.getStartPosition() + mWindow.getNumRows())) {
        fillWindow(newPosition);
    }

    return true;
}

private void fillWindow(int requiredPos) {
    clearOrCreateWindow(getDatabase().getPath());

    try {
        if (mCount == NO_COUNT) {
            int startPos = DatabaseUtils.cursorPickFillWindowStartPosition(requiredPos, 0);
            mCount = mQuery.fillWindow(mWindow, startPos, requiredPos, true);
            mCursorWindowCapacity = mWindow.getNumRows();
            if (Log.isLoggable(TAG, Log.DEBUG)) {
                Log.d(TAG, "received count(*) from native_fill_window: " + mCount);
            }
        } else {
            int startPos = DatabaseUtils.cursorPickFillWindowStartPosition(requiredPos,
                    mCursorWindowCapacity);
            mQuery.fillWindow(mWindow, startPos, requiredPos, false);
        }
    } catch (RuntimeException ex) {
        // Close the cursor window if the query failed and therefore will
        // not produce any results.  This helps to avoid accidentally leaking
        // the cursor window if the client does not correctly handle exceptions
        // and fails to close the cursor.
        closeWindow();
        throw ex;
    }
}

核心逻辑在mQuery.fillWindow(mWindow, startPos, requiredPos, false);这里。mQuery就是之前传入的SQLiteQuery对象。查看其fillWindow方法：

int fillWindow(CursorWindow window, int startPos, int requiredPos, boolean countAllRows) {
    acquireReference();
    try {
        window.acquireReference();
        try {
            int numRows = getSession().executeForCursorWindow(getSql(), getBindArgs(),
                    window, startPos, requiredPos, countAllRows, getConnectionFlags(),
                    mCancellationSignal);
            return numRows;
        } catch (SQLiteDatabaseCorruptException ex) {
            onCorruption();
            throw ex;
        } catch (SQLiteException ex) {
            Log.e(TAG, "exception: " + ex.getMessage() + "; query: " + getSql());
            throw ex;
        } finally {
            window.releaseReference();
        }
    } finally {
        releaseReference();
    }
}

可以看到，最终回到了SQLiteSession.executeXXX方法逻辑之下。其余即与上一节类似。

而从Cursor中取出数据的过程，则最终是由CursorWindow下的一系列native方法来完成，我认为属于Cursor的代码体系了，这里不重点展开。

5 Transaction

5.1 beginTransaction

//一群差不多的beginTransaction方法最终调用到了这里
private void beginTransaction(SQLiteTransactionListener transactionListener,
        boolean exclusive) {
    acquireReference();//怎么老是你
    try {
        getThreadSession().beginTransaction(
                exclusive ? SQLiteSession.TRANSACTION_MODE_EXCLUSIVE :
                        SQLiteSession.TRANSACTION_MODE_IMMEDIATE,
                transactionListener,
                getThreadDefaultConnectionFlags(false /*readOnly*/), null);
    } finally {
        releaseReference();
    }
}

//上面的方法调用了这个方法。这套flags做了两件小事：1.确定只读还是可写  2.如果是主线程，就要提高连接的优先级
int getThreadDefaultConnectionFlags(boolean readOnly) {
    int flags = readOnly ? SQLiteConnectionPool.CONNECTION_FLAG_READ_ONLY :
            SQLiteConnectionPool.CONNECTION_FLAG_PRIMARY_CONNECTION_AFFINITY;
    if (isMainThread()) {
        flags |= SQLiteConnectionPool.CONNECTION_FLAG_INTERACTIVE;
    }
    return flags;
}

还是要看SQLiteSession内部：

public void beginTransaction(int transactionMode,
        SQLiteTransactionListener transactionListener, int connectionFlags,
        CancellationSignal cancellationSignal) {
    throwIfTransactionMarkedSuccessful();//一点合法性检查，不贴了
    beginTransactionUnchecked(transactionMode, transactionListener, connectionFlags,
            cancellationSignal);
}

private void beginTransactionUnchecked(int transactionMode,
        SQLiteTransactionListener transactionListener, int connectionFlags,
        CancellationSignal cancellationSignal) {
    if (cancellationSignal != null) {
    //cancellationSignal从beginTransaction以及SQLiteStatement诸方法传入的均为null，调查发现仅query时可以传入此参数。
        cancellationSignal.throwIfCanceled();
    }

    if (mTransactionStack == null) {//Transaction栈为空时才获取连接。
        acquireConnection(null, connectionFlags, cancellationSignal); // might throw
    }
    try {
        // Set up the transaction such that we can back out safely
        // in case we fail part way.
        if (mTransactionStack == null) {//如果没有进行中的Transaction，创建一个并BEGIN
            // Execute SQL might throw a runtime exception.
            switch (transactionMode) {
                case TRANSACTION_MODE_IMMEDIATE:
                    mConnection.execute("BEGIN IMMEDIATE;", null,
                            cancellationSignal); // might throw
                    break;
                case TRANSACTION_MODE_EXCLUSIVE:
                    mConnection.execute("BEGIN EXCLUSIVE;", null,
                            cancellationSignal); // might throw
                    break;
                default:
                    mConnection.execute("BEGIN;", null, cancellationSignal); // might throw
                    break;
            }
        }

        // Listener might throw a runtime exception.
        if (transactionListener != null) {
            try {
                transactionListener.onBegin(); // might throw
            } catch (RuntimeException ex) {
                if (mTransactionStack == null) {
                    mConnection.execute("ROLLBACK;", null, cancellationSignal); // might throw
                }
                throw ex;
            }
        }

        // Bookkeeping can't throw, except an OOM, which is just too bad...
        Transaction transaction = obtainTransaction(transactionMode, transactionListener);//创建事务
        transaction.mParent = mTransactionStack;
        mTransactionStack = transaction;//入栈
    } finally {
        if (mTransactionStack == null) {//这里要栈为空时才释放连接。不为空时永远持有一个连接。
            releaseConnection(); // might throw
        }
    }
}

private static final class Transaction {
    public Transaction mParent;//这个是个链表，或者说在这里充当了一个栈
    public int mMode;
    public SQLiteTransactionListener mListener;
    public boolean mMarkedSuccessful;
    public boolean mChildFailed;
}

5.2 setTransactionSuccessful与endTransaction

直接看SQLiteSession吧：

public void setTransactionSuccessful() {
    throwIfNoTransaction();
    throwIfTransactionMarkedSuccessful();

    mTransactionStack.mMarkedSuccessful = true;//仅仅是个标记
}

public void endTransaction(CancellationSignal cancellationSignal) {
    throwIfNoTransaction();
    assert mConnection != null;

    endTransactionUnchecked(cancellationSignal, false);
}

private void endTransactionUnchecked(CancellationSignal cancellationSignal, boolean yielding) {
    if (cancellationSignal != null) {
        cancellationSignal.throwIfCanceled();
    }

    final Transaction top = mTransactionStack;
    boolean successful = (top.mMarkedSuccessful || yielding) && !top.mChildFailed;//如果有子Transaction失败，也是失败的

    RuntimeException listenerException = null;
    final SQLiteTransactionListener listener = top.mListener;
    if (listener != null) {
        try {
            if (successful) {
                listener.onCommit(); // might throw
            } else {
                listener.onRollback(); // might throw
            }
        } catch (RuntimeException ex) {
            listenerException = ex;
            successful = false;
        }
    }

    mTransactionStack = top.mParent;//退栈
    recycleTransaction(top);//回收

    if (mTransactionStack != null) {//还没到最外层事务，只做个标记
        if (!successful) {
            mTransactionStack.mChildFailed = true;
        }
    } else {//到了最外层事务了，提交或回滚
        try {
            if (successful) {
                mConnection.execute("COMMIT;", null, cancellationSignal); // might throw
            } else {
                mConnection.execute("ROLLBACK;", null, cancellationSignal); // might throw
            }
        } finally {
            releaseConnection(); // might throw
        }
    }

    if (listenerException != null) {
        throw listenerException;
    }
}

6 总结

(1)总的来说，SQLiteDatabase是线程安全且高效的。它并没有简单地对每次操作加锁，而是使用引用计数和ThreadLocal来保证连接复用的线程安全性，数据一致性则交由SQLite自身去保证，以达到最优性能。
而很多时候我们在业务层封装时反而处处加锁，其实是没有必要的。
(2)SQLiteDatabase的内部实现会让每个线程单独持有一个数据库连接（不一定是创建，因为有连接池优化），而不是每个SQLiteDatabase对象对应一个连接。
(3)数据库会给主线程持有的连接提高优先级。如果执行的是读操作或者小量数据的写入操作的话，可能可以满足主线程低延迟的需要。但是还没有具体的数据来支撑这一结论，希望有大牛补充。
(4)多线程下的事务行为本文中未作分析，下一篇会就此问题单独进行讨论。