如何在后台线程中从 Realm 复制结果，同时仍然侦听主线程的更改？

Question

java.lang.IllegalStateException: Realm access from incorrect thread. Realm objects can only be accessed on the thread they were created.

on

realm.copyFromRealm(results)

请参阅下面的代码片段 因为 copyFromRealm 很昂贵，我想在后台线程中进行。

注意：我无法在后台线程上获取 Realm 实例，因为 addChangeListener 函数需要在同一线程上运行。这个函数不能在没有looper的线程上调用，所以我必须使用主线程来添加监听器。因此，我需要想办法在Main线程中获取Realm实例，在Main线程中添加监听器，并在后台线程中执行copyFromRealm。

问题：如何在后台线程中从 Realm 复制结果，同时仍然监听主线程的更改？

代码：

override fun getUsersFlow(projectId: Int?): Flow<List<UserEntity>> {
return callbackFlow {

    val realm = Realm.getDefaultInstance()

    val projectBasedQuery = realm.where(UserEntity::class.java).apply {
        if (projectId != null) equalTo("projectId", projectId)
    }

    var copyJob: Job? = null
    val findAllAsync = projectBasedQuery.findAllAsync()

    findAllAsync.addChangeListener { results ->

        copyJob = CoroutineScope(Dispatchers.Default).launch {
            val items = results.realm.copyFromRealm(results)//CRASH
            val sortedItems = items.sortedByDescending { it.createdTime } // Sort updated list by date
            trySend(sortedItems)
        }
    }

    awaitClose {
        findAllAsync.removeAllChangeListeners()
        copyJob?.cancel()
        if (!realm.isClosed) {
            realm.close()
        }
    }

}.flowOn(Dispatchers.Main)
}

Answer 1

解决方案：

freeze()函数是关键。在代码中查看它是如何使用的

freezedResults.realm.copyFromRealm(freezedResults)

override fun getUsersFlow(projectId: Int?): Flow<List<UserEntity>> {
return callbackFlow {

    val realm = Realm.getDefaultInstance()

    val projectBasedQuery = realm.where(UserEntity::class.java).apply {
        if (projectId != null) equalTo("projectId", projectId)
    }

    var copyJob: Job? = null
    val findAllAsync = projectBasedQuery.findAllAsync()

    findAllAsync.addChangeListener { results ->

        val freezedResults = results.freeze() //creates new realm object available for reading in the background thread

        copyJob = CoroutineScope(Dispatchers.Default).launch {
            val items = freezedResults.realm.copyFromRealm(freezedResults) //use freezed instance to get data in the background thread
            val sortedItems = items.sortedByDescending { it.createdTime } // Sort updated list by date
            trySend(sortedItems)
        }
    }

    awaitClose {
        findAllAsync.removeAllChangeListeners()
        copyJob?.cancel()
        if (!realm.isClosed) {
            realm.close()
        }
    }

}.flowOn(Dispatchers.Main)
}

首选方法：

override fun getUsersFlow(projectId: Int?): Flow<List<UseEntity>> {
    // The use block ensures that the Realm instance is closed properly after its use
    return Realm.getDefaultInstance().use { realm ->
        realm.where(UseEntity::class.java).apply {
            if (projectId != null) equalTo("projectId", projectId)
        }.findAllAsync()
          .toFlow() // Converts the Realm results to Kotlin Flow
          .flowOn(Dispatchers.Main)
          .map { results ->
              // Copy results from Realm from the background thread
              realm.copyFromRealm(results)
                  .sortedByDescending { it.createdTime }
          }
          .flowOn(Dispatchers.Default)
    }
}

• 它在内部使用相同的 freeze() 函数。
• 确保完成后关闭领域。这可以使用 use 块来完成。
• 不用担心关闭 Realm 时流量会停止。 toFlow() 函数负责打开自己的 Realm 实例并在必要时关闭它。因此，您可以使用您的 Realm 创建流程，然后关闭您的 Realm，流程仍将继续。

重构：

fun <T : RealmModel, R> getRealmFlow(
   realmClass: Class<T>,
   queryBlock: (RealmQuery<T>) -> RealmQuery<T> = { it },
   mapAsync: (RealmResults<T>) -> R
): Flow<R> {
   return Realm.getDefaultInstance().use { defaultInstance ->
       defaultInstance.where(realmClass)
           .let(queryBlock)
           .findAllAsync()
           .toFlow().flowOn(Dispatchers.Main)
           .map { mapAsync(it) }
           .flowOn(Dispatchers.Default)
   }
}

override fun getUsersFlow(projectId: Int?): Flow<List<UserEntity>> {
   return getRealmFlow(
       realmClass = UserEntity::class.java,
       queryBlock = { query -> projectId?.let { query.equalTo("projectId", it) } ?: query },
       mapAsync = { it.copyFromRealm(it.realm).sortedByDescending { entity -> entity.createdTime } }
   )
}