Kotlin -使用协同技术实现Executors.newSingleThreadExecutor行为

Question

Executors.newSingleThreadExecutor将向其注册的任务排队，然后依次执行它们。以下代码：

val singleThreadedExecutor = Executors.newSingleThreadExecutor()

(0..10).forEach { i ->

    singleThreadedExecutor.execute {

        if (i % 2 == 0) {
            Thread.sleep(2000)
        } else {
            Thread.sleep(1000)
        }

        println(i)
    }
}

产出如下：

I/System.out: 0
I/System.out: 1
I/System.out: 2
I/System.out: 3
I/System.out: 4
I/System.out: 5
I/System.out: 6
I/System.out: 7
I/System.out: 8
I/System.out: 9
I/System.out: 10

我想利用科特林的合作机制来实现这种行为。我尝试过使用limitedParallelism，但它并没有像我预期的那样起作用。见下面的代码：

val singleThreadedCoroutine = Dispatchers.Default.limitedParallelism(1)

(0..10).forEach { i ->

    lifecycleScope.launch(singleThreadedCoroutine) {

        if (i % 2 == 0) {
            delay(2000)
        } else {
            delay(1000)
        }

        println(i)
    }
}

但其产出是：

I/System.out: 1
I/System.out: 3
I/System.out: 5
I/System.out: 7
I/System.out: 9
I/System.out: 0
I/System.out: 2
I/System.out: 4
I/System.out: 6
I/System.out: 8
I/System.out: 10

我是不是遗漏了什么？如何在协同线中对任务进行排队，使其按顺序执行？

谢谢。

Answer 1

这里的区别并不在于调度程序，而是在于挂起函数不阻塞线程，并且协同器是并发的。

delay()之所以被称为“非阻塞”，是因为它没有阻止从它调用的线程。这意味着线程可以在这段时间内自由地执行其他协同。另一方面，使用Thread.sleep()确实会阻塞线程，因此它将阻止线程在这段时间内执行其他任何操作，其他协同机制(或在您的情况下的任务)将不得不等待。如果您在协同方法中使用了Thread.sleep()，您应该会看到同样的结果，但这有点违背了协同工作的目的。

调度协同通常是按顺序进行的，但我不认为这是有文档记录的行为，而且很可能取决于调度人员。然而，在任何情况下，当它们暂停时，它们被允许交织(一般)--这几乎是并发的定义。

如果不需要并发性，则有几个选项：

在一个协同线中执行所有的工作:在一个launch中运行您的循环。

lifecycleScope.launch {
    repeat(11) { i ->

        if (i % 2 == 0) {
            delay(2000)
        } else {
            delay(1000)
        }

        println(i)
    }
}

1. 在一个协同线中完成所有工作，但在其他地方:使用Channel作为队列将事件发送到，而不是在每个项目上使用launch，只需通过通道发送项目即可。然后，生成一个单独的协同线，从通道中轮询元素以进行处理。

1. 如果您真正想要的只是保护某些部分不被并行运行，但是您可以使用并发性，否则，您也可以使用Mutex。在这种情况下，您甚至不必使用单线程分派程序(

)。

val mutex = Mutex()

repeat(11) { i ->

    lifecycleScope.launch {

        mutex.withLock {

            if (i % 2 == 0) {
                delay(2000)
            } else {
                delay(1000)
            }

            println(i)
        }
    }
}