diff --git a/app/src/main/java/io/legado/app/help/book/BookHelp.kt b/app/src/main/java/io/legado/app/help/book/BookHelp.kt index f70d93724..acbf32566 100644 --- a/app/src/main/java/io/legado/app/help/book/BookHelp.kt +++ b/app/src/main/java/io/legado/app/help/book/BookHelp.kt @@ -10,6 +10,7 @@ import io.legado.app.data.appDb import io.legado.app.data.entities.Book import io.legado.app.data.entities.BookChapter import io.legado.app.data.entities.BookSource +import io.legado.app.help.config.AppConfig import io.legado.app.model.analyzeRule.AnalyzeUrl import io.legado.app.model.localBook.LocalBook import io.legado.app.utils.ArchiveUtils @@ -24,12 +25,13 @@ import io.legado.app.utils.exists import io.legado.app.utils.externalFiles import io.legado.app.utils.getFile import io.legado.app.utils.isContentScheme +import io.legado.app.utils.onEachParallel import io.legado.app.utils.postEvent -import kotlinx.coroutines.Deferred import kotlinx.coroutines.Dispatchers.IO -import kotlinx.coroutines.async import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.collect +import kotlinx.coroutines.flow.flow import kotlinx.coroutines.withContext import org.apache.commons.text.similarity.JaccardSimilarity import splitties.init.appCtx @@ -151,19 +153,16 @@ object BookHelp { bookChapter: BookChapter, content: String ) = coroutineScope { - val awaitList = arrayListOf>() - val matcher = AppPattern.imgPattern.matcher(content) - while (matcher.find()) { - matcher.group(1)?.let { src -> + flow { + val matcher = AppPattern.imgPattern.matcher(content) + while (matcher.find()) { + val src = matcher.group(1) ?: continue val mSrc = NetworkUtils.getAbsoluteURL(bookChapter.url, src) - awaitList.add(async { - saveImage(bookSource, book, mSrc, bookChapter) - }) + emit(mSrc) } - } - awaitList.forEach { - it.await() - } + }.onEachParallel(AppConfig.threadCount) { mSrc -> + saveImage(bookSource, book, mSrc, bookChapter) + }.collect() } suspend fun saveImage( diff --git a/app/src/main/java/io/legado/app/model/webBook/BookChapterList.kt b/app/src/main/java/io/legado/app/model/webBook/BookChapterList.kt index fb7ccfdb4..240b010f7 100644 --- a/app/src/main/java/io/legado/app/model/webBook/BookChapterList.kt +++ b/app/src/main/java/io/legado/app/model/webBook/BookChapterList.kt @@ -11,13 +11,15 @@ import io.legado.app.data.entities.rule.TocRule import io.legado.app.exception.NoStackTraceException import io.legado.app.exception.TocEmptyException import io.legado.app.help.book.ContentProcessor +import io.legado.app.help.config.AppConfig import io.legado.app.model.Debug import io.legado.app.model.analyzeRule.AnalyzeRule import io.legado.app.model.analyzeRule.AnalyzeUrl import io.legado.app.utils.isTrue +import io.legado.app.utils.mapAsync import kotlinx.coroutines.Dispatchers.IO -import kotlinx.coroutines.async import kotlinx.coroutines.ensureActive +import kotlinx.coroutines.flow.flow import kotlinx.coroutines.withContext import splitties.init.appCtx import kotlin.coroutines.coroutineContext @@ -87,23 +89,23 @@ object BookChapterList { "◇并发解析目录,总页数:${chapterData.second.size}" ) withContext(IO) { - val asyncArray = Array(chapterData.second.size) { - async(IO) { - val urlStr = chapterData.second[it] - val res = AnalyzeUrl( - mUrl = urlStr, - source = bookSource, - ruleData = book, - headerMapF = bookSource.getHeaderMap() - ).getStrResponseAwait() //控制并发访问 - analyzeChapterList( - book, urlStr, res.url, - res.body!!, tocRule, listRule, bookSource, false - ).first + flow { + for (urlStr in chapterData.second) { + emit(urlStr) } - } - asyncArray.forEach { coroutine -> - chapterList.addAll(coroutine.await()) + }.mapAsync(AppConfig.threadCount) { urlStr -> + val res = AnalyzeUrl( + mUrl = urlStr, + source = bookSource, + ruleData = book, + headerMapF = bookSource.getHeaderMap() + ).getStrResponseAwait() //控制并发访问 + analyzeChapterList( + book, urlStr, res.url, + res.body!!, tocRule, listRule, bookSource, false + ).first + }.collect { + chapterList.addAll(it) } } } @@ -142,7 +144,8 @@ object BookChapterList { } } val replaceRules = ContentProcessor.get(book.name, book.origin).getTitleReplaceRules() - book.latestChapterTitle = list.last().getDisplayTitle(replaceRules, book.getUseReplaceRule()) + book.latestChapterTitle = + list.last().getDisplayTitle(replaceRules, book.getUseReplaceRule()) book.durChapterTitle = list.getOrElse(book.durChapterIndex) { list.last() } .getDisplayTitle(replaceRules, book.getUseReplaceRule()) if (book.totalChapterNum < list.size) { diff --git a/app/src/main/java/io/legado/app/service/ExportBookService.kt b/app/src/main/java/io/legado/app/service/ExportBookService.kt index 8ebab110c..6bb1cc0ca 100644 --- a/app/src/main/java/io/legado/app/service/ExportBookService.kt +++ b/app/src/main/java/io/legado/app/service/ExportBookService.kt @@ -42,6 +42,8 @@ import io.legado.app.utils.activityPendingIntent import io.legado.app.utils.cnCompare import io.legado.app.utils.createFolderIfNotExist import io.legado.app.utils.isContentScheme +import io.legado.app.utils.mapAsync +import io.legado.app.utils.mapAsyncIndexed import io.legado.app.utils.outputStream import io.legado.app.utils.postEvent import io.legado.app.utils.readBytes @@ -49,20 +51,14 @@ import io.legado.app.utils.readText import io.legado.app.utils.servicePendingIntent import io.legado.app.utils.toastOnUi import io.legado.app.utils.writeBytes -import kotlinx.coroutines.CoroutineStart import kotlinx.coroutines.Deferred -import kotlinx.coroutines.Dispatchers.Default import kotlinx.coroutines.Dispatchers.IO import kotlinx.coroutines.Job import kotlinx.coroutines.async import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.ensureActive -import kotlinx.coroutines.flow.buffer import kotlinx.coroutines.flow.collectIndexed import kotlinx.coroutines.flow.flow -import kotlinx.coroutines.flow.map -import kotlinx.coroutines.flow.onEach -import kotlinx.coroutines.flow.withIndex import kotlinx.coroutines.launch import me.ag2s.epublib.domain.Author import me.ag2s.epublib.domain.Date @@ -327,23 +323,19 @@ class ExportBookService : BaseService() { val threads = if (AppConfig.parallelExportBook) { AppConst.MAX_THREAD } else { - 0 + 1 } flow { appDb.bookChapterDao.getChapterList(book.bookUrl).forEach { chapter -> - val task = async(Default, start = CoroutineStart.LAZY) { - getExportData(book, chapter, contentProcessor, useReplace) - } - emit(task) - } - }.onEach { it.start() } - .buffer(threads) - .map { it.await() } - .collectIndexed { index, result -> - postEvent(EventBus.EXPORT_BOOK, book.bookUrl) - exportProgress[book.bookUrl] = index - append.invoke(result.first, result.second) + emit(chapter) } + }.mapAsync(threads) { chapter -> + getExportData(book, chapter, contentProcessor, useReplace) + }.collectIndexed { index, result -> + postEvent(EventBus.EXPORT_BOOK, book.bookUrl) + exportProgress[book.bookUrl] = index + append.invoke(result.first, result.second) + } } @@ -641,56 +633,52 @@ class ExportBookService : BaseService() { val threads = if (AppConfig.parallelExportBook) { AppConst.MAX_THREAD } else { - 0 + 1 } flow { - appDb.bookChapterDao.getChapterList(book.bookUrl).forEachIndexed { index, chapter -> - val task = async(Default, start = CoroutineStart.LAZY) { - val content = BookHelp.getContent(book, chapter) - val (contentFix, resources) = fixPic( - book, - content ?: if (chapter.isVolume) "" else "null", - chapter - ) - // 不导出vip标识 - chapter.isVip = false - val content1 = contentProcessor - .getContent( - book, - chapter, - contentFix, - includeTitle = false, - useReplace = useReplace, - chineseConvert = false, - reSegment = false - ).toString() - val title = chapter.run { - // 不导出vip标识 - isVip = false - getDisplayTitle( - contentProcessor.getTitleReplaceRules(), - useReplace = useReplace - ) - } - val chapterResource = ResourceUtil.createChapterResource( - title.replace("\uD83D\uDD12", ""), - content1, - contentModel, - "Text/chapter_${index}.html" - ) - ExportChapter(title, chapterResource, resources) - } - emit(task) + appDb.bookChapterDao.getChapterList(book.bookUrl).forEach { chapter -> + emit(chapter) } - }.onEach { it.start() } - .buffer(threads) - .map { it.await() } - .collectIndexed { index, exportChapter -> - postEvent(EventBus.EXPORT_BOOK, book.bookUrl) - exportProgress[book.bookUrl] = index - epubBook.resources.addAll(exportChapter.resources) - epubBook.addSection(exportChapter.title, exportChapter.chapterResource) + }.mapAsyncIndexed(threads) { index, chapter -> + val content = BookHelp.getContent(book, chapter) + val (contentFix, resources) = fixPic( + book, + content ?: if (chapter.isVolume) "" else "null", + chapter + ) + // 不导出vip标识 + chapter.isVip = false + val content1 = contentProcessor + .getContent( + book, + chapter, + contentFix, + includeTitle = false, + useReplace = useReplace, + chineseConvert = false, + reSegment = false + ).toString() + val title = chapter.run { + // 不导出vip标识 + isVip = false + getDisplayTitle( + contentProcessor.getTitleReplaceRules(), + useReplace = useReplace + ) } + val chapterResource = ResourceUtil.createChapterResource( + title.replace("\uD83D\uDD12", ""), + content1, + contentModel, + "Text/chapter_${index}.html" + ) + ExportChapter(title, chapterResource, resources) + }.collectIndexed { index, exportChapter -> + postEvent(EventBus.EXPORT_BOOK, book.bookUrl) + exportProgress[book.bookUrl] = index + epubBook.resources.addAll(exportChapter.resources) + epubBook.addSection(exportChapter.title, exportChapter.chapterResource) + } } data class ExportChapter( diff --git a/app/src/main/java/io/legado/app/utils/FlowExtensions.kt b/app/src/main/java/io/legado/app/utils/FlowExtensions.kt index b30152866..5a173252e 100644 --- a/app/src/main/java/io/legado/app/utils/FlowExtensions.kt +++ b/app/src/main/java/io/legado/app/utils/FlowExtensions.kt @@ -1,17 +1,22 @@ package io.legado.app.utils import kotlinx.coroutines.ExperimentalCoroutinesApi +import kotlinx.coroutines.async import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.channelFlow import kotlinx.coroutines.flow.filterNotNull import kotlinx.coroutines.flow.flatMapMerge import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.flow.map +import kotlinx.coroutines.flow.onEach +import kotlinx.coroutines.sync.Semaphore @OptIn(ExperimentalCoroutinesApi::class) inline fun Flow.onEachParallel( concurrency: Int, crossinline action: suspend (T) -> Unit ): Flow = flatMapMerge(concurrency) { value -> - return@flatMapMerge flow { + flow { action(value) emit(value) } @@ -37,3 +42,98 @@ inline fun Flow.onEachIndexed( emit(value) } } + +inline fun Flow.mapIndexed( + crossinline action: suspend (index: Int, T) -> R, +): Flow = flow { + var index = 0 + collect { value -> + emit(action(index++, value)) + } +} + +inline fun Flow.mapAsync( + concurrency: Int, + crossinline transform: suspend (T) -> R +): Flow = if (concurrency == 1) { + map { transform(it) } +} else { + Semaphore(concurrency).let { semaphore -> + channelFlow { + collect { + semaphore.acquire() + send(async { transform(it) }) + } + }.map { + it.await() + }.onEach { semaphore.release() } + } +} + +inline fun Flow.mapAsyncIndexed( + concurrency: Int, + crossinline transform: suspend (index: Int, T) -> R +): Flow = if (concurrency == 1) { + mapIndexed { index, value -> + transform(index, value) + } +} else { + Semaphore(concurrency).let { semaphore -> + channelFlow { + var index = 0 + collect { + semaphore.acquire() + val i = index++ + send(async { transform(i, it) }) + } + }.map { + it.await() + }.onEach { semaphore.release() } + } +} + +inline fun Flow.onEachAsync( + concurrency: Int, + crossinline action: suspend (T) -> Unit +): Flow = if (concurrency == 1) { + onEach { action(it) } +} else { + Semaphore(concurrency).let { semaphore -> + channelFlow { + collect { + semaphore.acquire() + send(async { + action(it) + it + }) + } + }.map { + it.await() + }.onEach { semaphore.release() } + } +} + +inline fun Flow.onEachAsyncIndexed( + concurrency: Int, + crossinline action: suspend (index: Int, T) -> Unit +): Flow = if (concurrency == 1) { + onEachIndexed { index, value -> + action(index, value) + } +} else { + Semaphore(concurrency).let { semaphore -> + channelFlow { + var index = 0 + collect { + semaphore.acquire() + val i = index++ + send(async { + action(i, it) + it + }) + } + }.map { + it.await() + }.onEach { semaphore.release() } + } +}