Jetpack:Room+kotlin协程? 事务问题分析，withTransaction API 详解.

刺骨的言语ヽ痛彻心扉 2022-09-11 11:10 142阅读 0赞

该文章接上一篇，[Room超详细使用踩坑指南][Room]。如果不了解Room的使用，可以先行参考上一篇文章。

### 文章目录 ###

*  问题简单说明
 *  简单解决
 *  room扩展方法
 *  withTransaction原理解析

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

# 问题简单说明 #

从Room 2.1 版本之后，可以定义**suspend** Dao来使用Kotlin协程了，如下所示：

@Insert
        suspend fun suspendInsertStudent(studentEntity: SimpleStudentEntity)
        @Query("select * from $SIMPLE_STUDENT_TABLE_NAME where $SIMPLE_STUDENT_TABLE_STUDENT_NAME = :name")
        suspend fun suspendStudentByName(name: String): List<SimpleStudentEntity>

在ViewModel中，可以简单的这么使用，如下所示：

suspend fun suspendUseTest(name : String){ 
            return withContext(Dispatchers.IO){ 
                simpleStudentDao.suspendStudentByName(name)
            }
        }

但是，**加入事务处理之后，就会有问题了，看下面一段代码**：

suspend fun transactionUseError(studentEntity: SimpleStudentEntity): List<SimpleStudentEntity> { 
            //使用了 IO dispatcher，所以该 DB 的操作在 IO 线程上进行
            return withContext(Dispatchers.IO) { 
                val dataBase = SimpleMyDataBase.getDataBase()
                //在 IO-Thread-1 线程上开始执行事务
                dataBase.beginTransaction()
                return@withContext try { 
                    // 协程可以在与调度器（这里就是 Dispatchers.IO）相关联的任何线程上绑定并继续执行。同时，由于事务也是在 IO-Thread-1 中开始的，因此我们可能恰好可以成功执行查询。
                    simpleStudentDao.suspendInsertStudent(studentEntity)//挂起函数
                    // 如果协程又继续在 IO-Thread-2 上执行，那么下列操作数据库的代码可能会引起死锁，因为它需要等到 IO-Thread-1 的线程执行结束后才可以继续。
                    simpleStudentDao.suspendStudentByName(studentEntity.name!!).apply { 
                        dataBase.setTransactionSuccessful() //永远不会执行这一行
                    }
                } finally { 
                    dataBase.endTransaction()//永远不会执行这一行
                    SimpleStudentEntity(name = null, age = null)
                }
            }
        }

Android 的 **SQLite 事务是受制于单个线程的。当一个正在进行的事务中的某个查询在当前线程中被执行时，它会被视为是该事务的一部分并允许继续执行。但当这个查询在另外一个线程中被执行时，那它就不再属于这个事务的一部分了，这样的话就会导致这个查询被阻塞，直到事务在另外一个线程执行完成**。

# 简单解决 #

其实可以简单封装一个api，去处理这个问题，直接限制使用一个线程即可，如下所示：

//简单封装
    suspend fun <T> RoomDatabase.simpleSolveTransactionUseError(block: suspend () -> T): T { 
        return withContext(newSingleThreadContext("sqlite")) { 
            beginTransaction()
            try { 
                return@withContext block()
            } finally { 
                endTransaction()
            }
        }
    }
    //使用
    suspend fun simpleSolveTransactionUseError(studentEntity: SimpleStudentEntity): List<SimpleStudentEntity> { 
        return SimpleMyDataBase.getDataBase().simpleSolveTransactionUseError { 
            simpleStudentDao.suspendInsertStudent(studentEntity)
            simpleStudentDao.suspendStudentByName(studentEntity.name!!)
        }
    }

但是，**这样的话，简单使用并没有问题，但是如果在代码块中出现另一个调度器呢**？如下所示：

SimpleMyDataBase.getDataBase().simpleSolveTransactionUseError { 
    	withContext(Dispatchers.Default){ 
    		simpleStudentDao.suspendInsertStudent(studentEntity)
    		simpleStudentDao.suspendStudentByName(studentEntity.name!!)
    	}
    }

这样的话，又会出现上述的问题了，不能保证都在一个线程执行。一个较好的实现是，**允许标准的协程构造器launch、async、withContent等等，允许在里面被创建，只有数据库的操作调用到对应的事务线程**。

# room扩展方法 #

对于上述问题，**Room提供了withTransaction API去解决该问题**，使用如下：

//直接进行事务操作
    suspend fun transactionInsertGet(studentEntity: SimpleStudentEntity): List<SimpleStudentEntity> { 
        return SimpleMyDataBase.getDataBase().withTransaction { 
            simpleStudentDao.suspendInsertStudent(studentEntity)
            simpleStudentDao.suspendStudentByName(studentEntity.name!!)
        }
    }
    //也可以在withTransaction 内部启动标准的协程
        suspend fun transactionInsertGet(studentEntity: SimpleStudentEntity): List<SimpleStudentEntity> { 
            return SimpleMyDataBase.getDataBase().withTransaction { 
                withContext(Dispatchers.Default) { 
                    //todo 其他的挂起操作 会调用到Default线程池
                    ...
                    //数据库操作 会被调用到专门的事务线程
                    simpleStudentDao.suspendInsertStudent(studentEntity)
                    simpleStudentDao.suspendStudentByName(studentEntity.name!!)
                }
            }
        }

API使用起来很简单，那么到底是如何实现的呢？我们这里进行深入分析一下。

# withTransaction原理解析 #

简单分析一下withTransaction都需要做到什么事情：

1.  单线程调度器，执行数据库操作
2.  记录事务当前的处理线程
3.  上下文需要可以判断当前Dao是否在事务中

首先保证，需要有一个地方记录专门处理数据库的线程。withTransaction启用runBlocking事件循环来获取线程的控制权。原代码如下：

private suspend fun Executor.acquireTransactionThread(controlJob: Job): ContinuationInterceptor { 
        return suspendCancellableCoroutine {  continuation ->
            continuation.invokeOnCancellation { 
                controlJob.cancel()
            }
            try { 
                execute { 
                    // runBlocking 创建一个 event loop 来执行协程中的任务代码
                    runBlocking { 
                        // Thread acquired, resume coroutine.
                        // 获取到线程后，通过返回有 runBlocking 创建的调度器。
                        continuation.resume(coroutineContext[ContinuationInterceptor]!!)
                        //挂起runBlocking直到controlJob结束
                        controlJob.join()
                    }
                }
            } catch (ex: RejectedExecutionException) { 
                // Couldn't acquire a thread, cancel coroutine.
                continuation.cancel(
                    IllegalStateException(
                        "Unable to acquire a thread to perform the database transaction.", ex
                    )
                )
            }
        }
    }

上述代码是Executor的扩展函数，**当线程获取到后，会执行execute函数，此时开启一个runBlocking，并且返回runBlocking的调度器（在后面的代码中，会将该调度器存在协程上下文当中），挂起runBlockiing直到结束（说明一点，runBlocking是在execute中执行的，所以所有通过runBlocking调度器调度的挂起函数都会在该线程被执行）**，如果当前没有可以使用的executor则挂起，如果有则该线程一直进入runBlocking的事件循环中，执行事务。

有了调度器之后，**就可以创建一个CoroutineContext.Element作为上下文来保存调度器了，如果在事务作用域内调用了 DAO 函数，就可以把 DAO 函数重新路由到相应的线程中**。源码如下：

internal class TransactionElement(
        private val transactionThreadControlJob: Job,
        internal val transactionDispatcher: ContinuationInterceptor
    ) : CoroutineContext.Element { 
    
        companion object Key : CoroutineContext.Key<TransactionElement>
    
        override val key: CoroutineContext.Key<TransactionElement>
            get() = TransactionElement
    
        /** *这个 element 用来统计事务数量（包含嵌套事务）。调用 [acquire]来增加计数， *调用 [release] 来减少计数。如果在调用 [release] 时计数达到 0，则事务被取消，事务线程会被释放。 */
        private val referenceCount = AtomicInteger(0)
    
        fun acquire() { 
            referenceCount.incrementAndGet()
        }
    
        fun release() { 
            val count = referenceCount.decrementAndGet()
            if (count < 0) { 
                throw IllegalStateException("Transaction was never started or was already released.")
            } else if (count == 0) { 
                // Cancel the job that controls the transaction thread, causing it to be released.
                transactionThreadControlJob.cancel()
            }
        }
    }

TransactionElement 函数中的 acquire 和 release 是用来跟踪嵌套事务的。在最外层事务完成时释放事务线程即可。

另外事务线程需要标记目前是否进入了事务，所以，需要一个保存的容器，以线程为作用域的很容易就想到了ThreadLocal，在RoomDatabase类中保存了一个ThreadLocal，如下所示：

/** * This id is only set on threads that are used to dispatch coroutines within a suspending * database transaction. */
        private final ThreadLocal<Integer> mSuspendingTransactionId = new ThreadLocal<>();

上文中提到的创建事务上下文中所需的最后一个关键元素是 `ThreadContextElement`。CoroutineContext 中的这个元素类似于 `ThreadLocal`，它能够跟踪线程中是否有正在进行的事务。它就是获取的`mSuspendingTransactionId`然后转成`context`的。

suspendingTransactionId.asContextElement(System.identityHashCode(controlJob))

这样，事务的上下文元素就可以创建了，如下所示：

private suspend fun RoomDatabase.createTransactionContext(): CoroutineContext { 
        val controlJob = Job()
        coroutineContext[Job]?.invokeOnCompletion { 
            controlJob.cancel()
        }
        val dispatcher = transactionExecutor.acquireTransactionThread(controlJob)
        val transactionElement = TransactionElement(controlJob, dispatcher)
        val threadLocalElement =
            suspendingTransactionId.asContextElement(System.identityHashCode(controlJob))
        return dispatcher + transactionElement + threadLocalElement
    }

**整个`withTransaction`API的实现**

public suspend fun <R> RoomDatabase.withTransaction(block: suspend () -> R): R { 
        // Use inherited transaction context if available, this allows nested suspending transactions.
        val transactionContext =
            coroutineContext[TransactionElement]?.transactionDispatcher ?: createTransactionContext()
        return withContext(transactionContext) { 
            val transactionElement = coroutineContext[TransactionElement]!!
            transactionElement.acquire()
            try { 
                @Suppress("DEPRECATION")
                beginTransaction()
                try { 
                    val result = block.invoke()
                    @Suppress("DEPRECATION")
                    setTransactionSuccessful()
                    return@withContext result
                } finally { 
                    @Suppress("DEPRECATION")
                    endTransaction()
                }
            } finally { 
                transactionElement.release()
            }
        }
    }

看到这里会不会有个疑问？\*\*怎么保证在withTransactionAPI里面继续调度，并且Dao还是还是执行在事务线程的呢？刚开始我也是反反复复研究了好久，最后返现，对应的suspend Dao函数里面也有操作，\*\*随便点一个suspend Dao看看吧：

CoroutinesRoom.execute(__db, true, new Callable<Unit>() { 
          @Override
          public Unit call() throws Exception { 
          ...
            }, p1);

里面调用了CoroutinesRoom.execute，继续找execute看看。

@JvmStatic
    public suspend fun <R> execute(
    	db: RoomDatabase,
    	inTransaction: Boolean,
    	cancellationSignal: CancellationSignal,
    	callable: Callable<R>
    ): R { 
    	if (db.isOpen && db.inTransaction()) { 
    		return callable.call()
    	}
    
    	// Use the transaction dispatcher if we are on a transaction coroutine, otherwise
    	// use the database dispatchers.
    	val context = coroutineContext[TransactionElement]?.transactionDispatcher
    		?: if (inTransaction) db.transactionDispatcher else db.queryDispatcher
    	return suspendCancellableCoroutine<R> {  continuation ->
    		val job = GlobalScope.launch(context) { 
    		try { 
    			val result = callable.call()
    			continuation.resume(result)
    		} catch (exception: Throwable) { 
    			continuation.resumeWithException(exception)
    		}
    	}
    	continuation.invokeOnCancellation { 
    		if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.JELLY_BEAN) { 
    			cancellationSignal.cancel()
    		}
    			job.cancel()
    		}
    	}
    }

**注意看`coroutineContext[TransactionElement]`，这个不就是存进去的runBlocking的调度器嘛，所以就会调度到对应的事务线程了。通过`TransactionElement` 函数中的 `acquire` 和 `release`可以保证Dao的嵌套！！！是不是很强**！

[Room]: https://blog.csdn.net/weixin_44235109/article/details/120252231?spm=1001.2014.3001.5501