Как присоединиться к Kotlin SupervisorJob

Question

Я пытаюсь обработать дерево объектов данных.Предполагается, что каждый лист дерева обрабатывается с помощью сопрограммы.Весь процесс должен выполняться с использованием пула потоков фиксированного размера.

Итак, я пришел к следующему:

val node = an instance of WorkspaceEntry (tree structure)
val localDispatcher = newFixedThreadPoolContext(16)

fun main() {
    val job = SupervisorJob()
    val scope = CoroutineScope(localDispatcher + job)
    handleEntry(node, scope)

    runBlocking {
        job.join()
    }
}

Метод handleEntry рекурсивно запускает дочернее задание в супервизоре для каждого листа дерева.

Все дочерние задания супервизора успешно завершены, но соединение никогда не возвращается.Я правильно понял?

Редактировать: функция HandleEntry

private fun handleEntry(workspaceEntry: WorkspaceEntry, scope: CoroutineScope) {
    if (workspaceEntry is FileEntry) {
        scope.launch {
            FileTypeRegistry.processFile(workspaceEntry.fileBlob)
        }
    } else {
        workspaceEntry.children.forEach { child -> handleEntry(child, scope) }
    }
}

Answer 1

Вы смешали две роли:

1. основное задание, найденное в области сопрограмм, которое никогда не завершается само по себе и используется для управления жизненным циклом всего остального
2. заданиясоответствующая единице работы, возможно, разложенная на более дочерние задания

Вам нужны оба варианта, например:

val masterJob = SupervisorJob()
val scope = CoroutineScope(localDispatcher + masterJob)

val unitOfWork = scope.launch { handleEntry(node, scope) }
runBlocking { unitOfWork.join() }

Приведенный выше код на самом деле не мотивирует существованиеосновная работа, потому что вы запускаете только одну дочернюю работу из нее, но это может иметь смысл в более широкой картине, где у вас есть некоторый контекст, из которого вы запускаете много работ, и хотите иметь возможность писать

masterJob.cancel()

отменить все, прежде чем это сделано.

Answer 2

Кажется, Job, который используется для создания CoroutineContext (в вашем случае SupervisorJob), не предназначен для ожидания завершения дочерних сопрограмм, поэтому вы не можете использовать job.join().Я предполагаю, что главная цель этого Job состоит в том, чтобы отменить дочерние сопрограммы.Изменение блока runBlocking на следующее будет работать:

runBlocking {
    job.children.forEach {
        it.join()
    }
}