Kotlin 用协程同步线程

Question

目前我有一个不断循环的主线程：

var suspension = Suspension()

fun loop() {
    // Doing stuff here...
        
    suspension.callTick()
        
    // Doing more stuff here...
}

它调用 callTick() 方法将数据发送到通道：

class Suspension {
    private val ticksChannel = Channel<Unit>(0)

    fun callTick() {
        ticksChannel.trySend(Unit)
    }

    suspend fun waitTick() {
        ticksChannel.receive()
    }
}

现在我的最后一堂课使用了这个：

class Task(private val suspension: Suspension) {
    suspend fun runTask() {
        while (true) {
            suspension.waitTick()

            someMethodOnTheMainThread()
        }
    }
}

现在我想知道如何从主线程调用 someMethodOnTheMainThread() 方法。该函数必须在 loop() 中的 'suspension.callTick()' 方法之后立即调用。目前我正在从协程线程运行该函数。这会导致很多错误和空指针异常，因为它没有与主线程同步。

基本上我想知道如何阻塞/锁定主线程，直到调用suspend.waitTick（）方法并运行它之后的代码。这太复杂了吗？还有其他方法可以使挂起函数与同步代码一起使用吗？

Answer 1

好的，所以我尝试为客户端线程实现我自己的调度程序，如下所示：

/**
 * Dispatches execution onto Client event dispatching thread and provides native [delay] support.
 */
@Suppress("unused")
val Dispatchers.Client : ClientDispatcher
    get() = Client

/**
 * Dispatcher for Client event dispatching thread.
 *
 * This class provides type-safety and a point for future extensions.
 */
@OptIn(InternalCoroutinesApi::class)
sealed class ClientDispatcher : MainCoroutineDispatcher(), Delay {
    override fun dispatch(context: CoroutineContext, block: Runnable): Unit = ClientUtility.invokeLater(block)

    @OptIn(ExperimentalCoroutinesApi::class)
    override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation<Unit>) {
        val timer = schedule(timeMillis, TimeUnit.MILLISECONDS, ActionListener {
            with(continuation) { resumeUndispatched(Unit) }
        })
        continuation.invokeOnCancellation { timer.stop() }
    }

    /** @suppress */
    override fun invokeOnTimeout(timeMillis: Long, block: Runnable, context: CoroutineContext): DisposableHandle {
        val timer = schedule(timeMillis, TimeUnit.MILLISECONDS, ActionListener {
            block.run()
        })
        return object : DisposableHandle {
            override fun dispose() {
                timer.stop()
            }
        }
    }

    private fun schedule(time: Long, unit: TimeUnit, action: ActionListener): Timer =
        Timer(unit.toMillis(time).coerceAtMost(Int.MAX_VALUE.toLong()).toInt(), action).apply {
            isRepeats = false
            start()
        }
}

@OptIn(InternalCoroutinesApi::class)
internal class ClientDispatcherFactory : MainDispatcherFactory {
    override val loadPriority: Int
        get() = 0

    override fun createDispatcher(allFactories: List<MainDispatcherFactory>): MainCoroutineDispatcher = Client
}

private object ImmediateClientDispatcher : ClientDispatcher() {
    override val immediate: MainCoroutineDispatcher
        get() = this

    override fun isDispatchNeeded(context: CoroutineContext): Boolean = !ClientUtility.isEventDispatchThread()

    @OptIn(InternalCoroutinesApi::class)
    override fun toString() = toStringInternalImpl() ?: "Client.immediate"
}

/**
 * Dispatches execution onto Client event dispatching thread and provides native [delay] support.
 */
internal object Client : ClientDispatcher() {
    override val immediate: MainCoroutineDispatcher
        get() = ImmediateClientDispatcher

    @OptIn(InternalCoroutinesApi::class)
    override fun toString() = toStringInternalImpl() ?: "Client"
}

实际上是Swing dispatcher。主要是我将 invokeLater 方法更改为我自己的实现。另一个值得注意的修改是我使用自己的 isEventDispatchThread 实现。

我仍在使用 Swing Timer 类。我不确定这是否可以...但它似乎可以通过一些测试来工作。

对于invokeLater和isEventDispatchThread的实现：

    val eventQueue = ConcurrentLinkedQueue<Runnable>()

    fun invokeLater(block: Runnable) {
        eventQueue.add(block)
    }

    fun isEventDispatchThread() = if (dispatchThread == null) {
            false
        } else {
            Thread.currentThread() == dispatchThread
        }

    var dispatchThread: Thread? = null

然后在循环方法中：

    fun loop() {
        // Doing stuff here

        suspension.callTick()

        ClientUtility.dispatchThread = Thread.currentThread()
        val iterator = ClientUtility.eventQueue.iterator()
        while (iterator.hasNext()) {
            val block = iterator.next()
            block.run()
            iterator.remove()
        }

        // Doing some more stuff here
    }

我现在启动协程：Dispatchers.Main。哪个调度程序是我在resources/META-INF/services.kotlinx.coroutines.internal.MainDispatcherFactory 文件中定义的主要调度程序。

Answer 2

在主线程调用someMethodOnTheMainThread()方法 并暂停当前协程，你可以像下面这样定义它：

suspend fun someMethodOnTheMainThread() = withContext(Dispatchers.Main) {
    // ... your code here
}

withContext - 更改协程的上下文并在主线程中运行代码。

如果我们使用 kotlinx-coroutines-android 工件，

Dispatchers.Main 可以在 Android 上使用。同样，在 JavaFX 上如果我们使用 kotlinx-coroutines-javafx，在 Swing 上如果我们使用 kotlinx-coroutines-swing。可能还有其他一些设置它的库。如果你不使用任何一个，这个dispatcher没有配置，不能使用，所以你需要自己创建dispatcher。

Answer 3

如果您没有使用控制主线程的框架（并且会提供主线程调度程序），您可以执行以下操作：

var mainScope: CoroutineScope? = null;

fun main() {
    runBlocking {
        mainScope = this
        while(true) {
            //...
            suspension.callTick()
            // allow other coroutines to run
            yield()
            //...
        }
    }
}

如果你想从其他地方在主线程上运行某些东西，你可以使用mainScope?.launch。