有没有办法使用协程获得 FixedTreadPool 行为？答案

【问题标题】：Is there way to get FixedTreadPool behavior using coroutines?有没有办法使用协程获得 FixedTreadPool 行为？
【发布时间】：2017-11-28 00:13:39
【问题描述】：

有没有办法获得与下面的代码 sn-p 相同的行为但使用协程？

更新代码 sn-p：

fun main(args: Array<String>) = runBlocking {
    val executor = Executors.newFixedThreadPool(50)
    log.info("Start")
    val jobs = List(300) {
        executor.submit {
            log.info("worker #$it started")
            sleep(1000L)
            log.info("worker #$it done")
        }
    }
    jobs.forEach { it.get() }
    executor.shutdown()
    log.info("All done!")
}

如何运行 300 个并行因子 == 50 的作业，但不创建 50 个实际线程？

更新 2：解决方案

再读一遍Coroutines Guide 后，我发现Fan-out example 正是我想要的。因此，我的示例如下所示：

fun produceTasks() = produce {
    for (taskId in 1..300) {
        send(
                async(start = CoroutineStart.LAZY) {
                    delay(1000) // simulate long work
                    taskId
                }
        )
    }
    close()
}

fun launchWorker(index: Int, channel: ProducerJob<Deferred<Int>>) = launch {
    channel.consumeEach {
        val result = it.await()
        log.info("Worker #$index done task #$result")
    }
}

fun main(args: Array<String>) = runBlocking {
    val tasks = produceTasks()
    val workers = List(50) { launchWorker(it + 1, tasks) }
    workers.forEach { it.join() }
    log.info("Done")
}

【问题讨论】：

当您说“300 个并行因子 == 50 的作业”时，如果您认为这不是多个底层真实线程，那么“并行因子”是什么意思？
我的意思是作业应该以长度 50 排队，但使用轻量级协程（可能有大约 4 个真正的线程底层而不是 50 个真正的线程）。但是，如果我在下面的评论中写下类似的内容，那么所有 300 个工作/任务都是同时开始的。

标签： kotlin kotlinx.coroutines

【解决方案1】：

首先-请check issue on Gitbub，可能有官方解决方案。

你可以为这个逻辑使用协程通道

/**
 * How it works:
 * 1. Executors queue has N items
 * 2. Before each execution:
 * 2.1. Get item from queue (if it is possible)
 * 2.2. Execute action
 * 2.3. Return item to queue
 *
 * Example below allows us to execute methods in the resticted count of threads.
 * However we can create object pool from this Semaphore:
 * 1. Firstly - push N pooled objects into executors variable (e.g. we will have Channel<PooledObject>
 * 2. Then change signature of schedule method: just change arg "()" to PooledObject
 */
class Semapshore(maxParallelWorkers: Int) {
    private val executors = Channel<Unit>(maxParallelWorkers).apply {
        runBlocking {
            repeat(maxParallelWorkers) {
                send(Unit)
            }
        }
    }

    suspend fun <TResult> schedule(func: () -> TResult): TResult {
        val executor = executors.receive()

        return try {
            func()
        } finally {
            executors.send(executor)
        }
    }
}

【讨论】：