优化 #5564

app/src/main/java/io/legado/app/model/CacheBook.kt

@@ -11,18 +11,29 @@ import io.legado.app.data.entities.BookSource
 import io.legado.app.exception.ConcurrentException
 import io.legado.app.help.book.BookHelp
 import io.legado.app.help.book.isLocal
+import io.legado.app.help.config.AppConfig
 import io.legado.app.help.coroutine.CompositeCoroutine
 import io.legado.app.help.coroutine.Coroutine
 import io.legado.app.model.webBook.WebBook
 import io.legado.app.service.CacheBookService
+import io.legado.app.utils.onEachParallel
 import io.legado.app.utils.postEvent
 import io.legado.app.utils.startService
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.Dispatchers.IO
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.currentCoroutineContext
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.Semaphore
+import kotlinx.coroutines.sync.withLock
 import java.util.concurrent.ConcurrentHashMap
 import kotlin.coroutines.CoroutineContext
 
@@ -30,6 +41,9 @@ object CacheBook {
 
     val cacheBookMap = ConcurrentHashMap<String, CacheBookModel>()
 
+    private val workingState = MutableStateFlow(true)
+    private val mutex = Mutex()
+
     @Synchronized
     fun getOrCreate(bookUrl: String): CacheBookModel? {
         val book = appDb.bookDao.getBook(bookUrl) ?: return null
@@ -104,6 +118,36 @@ object CacheBook {
         errorDownloadMap.clear()
     }
 
+    fun setWorkingState(value: Boolean) {
+        workingState.value = value
+    }
+
+    suspend fun startProcessJob(context: CoroutineContext) = mutex.withLock {
+        setWorkingState(true)
+        flow {
+            while (currentCoroutineContext().isActive && cacheBookMap.isNotEmpty()) {
+                var emitted = false
+
+                cacheBookMap.forEach { (_, model) ->
+                    if (!model.isLoading()) {
+                        emit(model)
+                        emitted = true
+                    }
+                    workingState.first { it }
+                }
+
+                if (!emitted) {
+                    delay(1000)
+                }
+            }
+        }.onEachParallel(AppConfig.threadCount) {
+            coroutineScope {
+                it.download(this, context)
+            }
+        }.collect()
+    }
+
+
     val downloadSummary: String
         get() {
             return "正在下载:${onDownloadCount}|等待中:${waitCount}|失败:${errorDownloadMap.count()}|成功:${successDownloadSet.size}"
@@ -111,11 +155,12 @@ object CacheBook {
 
     val isRun: Boolean
         get() {
-            var isRun = false
             cacheBookMap.forEach {
-                isRun = isRun || it.value.isRun()
+                if (it.value.isRun()) {
+                    return true
+                }
             }
-            return isRun
+            return false
         }
 
     private val waitCount: Int
@@ -146,6 +191,7 @@ object CacheBook {
         private val tasks = CompositeCoroutine()
         private var isStopped = false
         private var waitingRetry = false
+        private var isLoading = true
 
         val waitCount get() = waitDownloadSet.size
         val onDownloadCount get() = onDownloadSet.size
@@ -156,7 +202,7 @@ object CacheBook {
 
         @Synchronized
         fun isRun(): Boolean {
-            return waitDownloadSet.isNotEmpty() || onDownloadSet.isNotEmpty()
+            return waitDownloadSet.isNotEmpty() || onDownloadSet.isNotEmpty() || isLoading
         }
 
         @Synchronized
@@ -164,6 +210,11 @@ object CacheBook {
             return isStopped || (!isRun() && !waitingRetry)
         }
 
+        @Synchronized
+        fun isLoading(): Boolean {
+            return isLoading
+        }
+
         @Synchronized
         fun stop() {
             waitDownloadSet.clear()
@@ -181,6 +232,7 @@ object CacheBook {
                 }
             }
             cacheBookMap[book.bookUrl] = this
+            isLoading = false
             postEvent(EventBus.UP_DOWNLOAD, book.bookUrl)
         }
 
@@ -243,7 +295,7 @@ object CacheBook {
             postEvent(EventBus.UP_DOWNLOAD, book.bookUrl)
             val chapterIndex = waitDownloadSet.firstOrNull()
             if (chapterIndex == null) {
-                if (onDownloadSet.isEmpty()) {
+                if (!isLoading && onDownloadSet.isEmpty()) {
                     cacheBookMap.remove(book.bookUrl)
                 }
                 return

app/src/main/java/io/legado/app/service/CacheBookService.kt

@@ -161,23 +161,8 @@ class CacheBookService : BaseService() {
     private fun download() {
         downloadJob?.cancel()
         downloadJob = lifecycleScope.launch(cachePool) {
-            while (isActive) {
-                if (!CacheBook.isRun) {
-                    stopSelf()
-                    return@launch
-                }
-                CacheBook.cacheBookMap.forEach {
-                    val cacheBookModel = it.value
-                    while (cacheBookModel.waitCount > 0) {
-                        if (CacheBook.onDownloadCount < threadCount) {
-                            cacheBookModel.download(this, cachePool)
-                        } else {
-                            delay(100)
-                        }
-                    }
-                }
-                delay(100)
-            }
+            CacheBook.startProcessJob(cachePool)
+            stopSelf()
         }
     }
 

app/src/main/java/io/legado/app/ui/main/MainViewModel.kt

@@ -213,28 +213,14 @@ class MainViewModel(application: Application) : BaseViewModel(application) {
         if (AppConfig.preDownloadNum == 0) return
         cacheBookJob?.cancel()
         cacheBookJob = viewModelScope.launch(upTocPool) {
-            while (isActive) {
-                if (CacheBookService.isRun || !CacheBook.isRun) {
-                    cacheBookJob?.cancel()
-                    cacheBookJob = null
-                    return@launch
+            launch {
+                while (isActive && CacheBook.isRun) {
+                    //有目录更新是不缓存,优先更新目录,现在更多网站限制并发
+                    CacheBook.setWorkingState(waitUpTocBooks.isEmpty() && onUpTocBooks.isEmpty())
+                    delay(1000)
                 }
-                CacheBook.cacheBookMap.forEach {
-                    val cacheBookModel = it.value
-                    while (cacheBookModel.waitCount > 0) {
-                        //有目录更新是不缓存,优先更新目录,现在更多网站限制并发
-                        if (waitUpTocBooks.isEmpty()
-                            && onUpTocBooks.isEmpty()
-                            && CacheBook.onDownloadCount < threadCount
-                        ) {
-                            cacheBookModel.download(this, upTocPool)
-                        } else {
-                            delay(100)
-                        }
-                    }
-                }
-                delay(100)
             }
+            CacheBook.startProcessJob(upTocPool)
         }
     }
 

app/src/main/java/io/legado/app/utils/FlowExtensions.kt

@@ -5,6 +5,7 @@ import androidx.lifecycle.repeatOnLifecycle
 import io.legado.app.data.appDb
 import kotlinx.coroutines.ExperimentalCoroutinesApi
 import kotlinx.coroutines.async
+import kotlinx.coroutines.currentCoroutineContext
 import kotlinx.coroutines.ensureActive
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.FlowCollector
@@ -20,7 +21,6 @@ import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.flow.produceIn
 import kotlinx.coroutines.sync.Semaphore
-import kotlin.coroutines.coroutineContext
 
 @OptIn(ExperimentalCoroutinesApi::class)
 inline fun <T> Flow<T>.onEachParallel(
@@ -42,7 +42,7 @@ inline fun <T> Flow<T>.onEachParallelSafe(
         try {
             action(value)
         } catch (e: Throwable) {
-            coroutineContext.ensureActive()
+            currentCoroutineContext().ensureActive()
         }
         emit(value)
     }
@@ -64,7 +64,7 @@ inline fun <T, R> Flow<T>.mapParallelSafe(
         try {
             emit(transform(value))
         } catch (_: Throwable) {
-            coroutineContext.ensureActive()
+            currentCoroutineContext().ensureActive()
         }
     }
 }.buffer(0)
@@ -78,7 +78,7 @@ inline fun <T, R> Flow<T>.transformParallelSafe(
         try {
             transform(value)
         } catch (e: Throwable) {
-            coroutineContext.ensureActive()
+            currentCoroutineContext().ensureActive()
         }
     }
 }.buffer(0)