在 Kotlin 中正确实现等待和通知

Question

根据本文档，不鼓励在 Kotlin 中使用 wait 和 notify：https://kotlinlang.org/docs/reference/java-interop.html

等待()/通知()

Effective Java Item 69 建议优先使用并发实用程序而不是 wait() 和 notify()。因此，这些方法不适用于 Any 类型的引用。

但是该文档没有提出任何正确的方法。

基本上，我想实现一个服务，它会读取输入数据并处理它们。如果没有输入数据，它将暂停自己，直到有人通知有新的输入数据。类似的东西

while (true) {
    val data = fetchData()
    processData(data)
    if (data.isEmpty()) {
        wait()
    }
}

编辑：

我不想使用这些不推荐的方法（反模式），我真的很想知道如何正确地做到这一点。

在我的情况下，fetchData 从数据库中读取数据，因此在我的情况下无法使用队列。

Answer 1

一般来说，您应该尽可能使用更高级别的并发实用程序。

但是，如果在您的情况下没有更高级别的构造起作用，则直接 替换是使用 ReentrantLock 和单 Condition 在那把锁上。

例如，如果您的 Java 代码类似于：

private Object lock = new Object();

...

synchronized(lock) {
    ...
    lock.wait();
    ...
    lock.notify();
    ...
    lock.notifyAll();
    ...
}

您可以将其更改为以下 Kotlin：

private val lock = ReentrantLock()
private val condition = lock.newCondition()

lock.withLock {           // like synchronized(lock)
    ...
    condition.await()     // like wait()
    ...
    condition.signal()    // like notify()
    ...
    condition.signalAll() // like notifyAll()
    ...
}

虽然这有点冗长，但条件确实提供了一些额外的 灵活性，因为您可以在单个锁上具有多个条件，并且 还有其他类型的锁（特别是ReentrantReadWriteLock.ReadLock 和 ReentrantReadWriteLock.WriteLock)。

请注意，withLock 是 Kotlin 提供的扩展函数，负责在调用提供的 lambda 之前/之后调用 Lock.lock()/Lock.unlock()。

Answer 2

BlockingQueue 可以是适合您用例的高级并发实用程序，但应用它需要了解和修改您的代码结构。

这个想法是，fetchData() 应该 .take() 队列中的一个项目，如果队列为空，则会阻止执行，直到出现项目，这消除了代码中的 .wait()。数据的生产者应该.put(t) 将数据放入队列。

---

如果您真的需要使用wait 和notify，例如为了在低级别实现并发实用程序，您可以将 Kotlin 对象强制转换为 java.lang.Object 并在之后调用这些函数，如 the language reference 中所述。或者，写成扩展函数：

@Suppress("PLATFORM_CLASS_MAPPED_TO_KOTLIN")
private fun Any.wait() = (this as java.lang.Object).wait()

Answer 3

您可以使用kotlin Semaphore 来暂停协程而不是阻塞线程：

val semaphore = Semaphore(1, 0)

suspend fun runFetchLoop() = withContext(Dispatchers.Default){
    while (isActive) {
        val data = fetchData()
        processData(data)
        semaphore.acquire()
        if (data.isEmpty()) {
            semaphore.acquire()
        }
        semaphore.release()
    }
}

fun notifyDataAvailable() = semaphore.release()

这是一个运行示例：https://pl.kotl.in/bnKeOspfs。

但是我更喜欢 cold Flow 或 hot Channel 来解决这个问题。这是一个很好的article，关于伟大的 Roman Elizarov 的冷流和热通道