如何使用 ReentrantLock 或 Mutex 正确锁定线程？

Question

我有一个带有回调的 acync 方法：

myAsyncMethod(params) {
    handleResult(it)
}

我需要将其转换为同步方法（阻止并返回结果），如下所示：

val result = mySyncMethod(params)
handleResult(result)

因为这个方法是第三方库提供的，所以无法重新实现。

我正在尝试使用 ReentrantLock：

fun mySyncMethod(params:Type) {
    Log.d("tag", "1")
    val result = null
    val mutex = ReentrantLock()
    myAsyncMethod(params) {
        Log.d("tag", "3")
        result = it
        mutex.unlock()
    }
    Log.d("tag", "2")
    mutex.lock()
    Log.d("tag", "4")
    return result
}

handleResult(mySyncMethod(params))

我等待我应该看到 1、2、3、4。但我在 handleResult 中得到 1、2、4、3 和 null。我正在尝试使用互斥锁，但结果相同。如何让它发挥作用？

附：当然我可以使用同步，但在这种情况下我必须另外使用对象变量。

Answer 1

这是我的代码。我用线程代替你的方法来测试自己。

fun main() {
    val mutex = ReentrantLock(true)
    val asyncCondition = mutex.newCondition()
    mutex.lock()
    try
    {
        print(1)


        thread {
            mutex.lock()
            try
            {
                print(2)
            }
            finally
            {
                asyncCondition.signalAll()
                mutex.unlock()
            }
        }

        asyncCondition.await()
        print(3)
    }
    finally
    {
        mutex.unlock()
    }
}

Java 的ReentrnatLock.lock 方法将在一个线程已经锁定在同一个线程上时阻塞。由于您试图从另一个线程锁定异步线程，因此您的代码不起作用。要阻止或释放另一个线程，您可以像我的代码一样使用Condition。

我之所以到处使用try..finallystatement，是因为如果锁定线程后发生异常，就会遇到无限循环，所以一定要小心这种情况。

您可能会看到这些参考资料。

• Concurrency (ReentrantLock) in different threads
• How do java.util.concurrent.locks.Condition work?

Answer 2

我希望我应该看到 1、2、3、4。

那你误解了ReentrantLock 的实际用途。

如果mySyncMethod可以从不同的线程调用，那么在调用mutex.lock()之后，一次只能有一个线程可以执行mutex.lock()之后的代码——其他执行该方法的线程必须等待。一旦调用mutex.unlock()，其中一个等待线程也将执行代码，依此类推。

因此，拥有mutex 的线程会强制其他线程等待，但它自己的流程不会被阻塞。 myAsyncMethod 在外部方法完成后开始运行，因此 result 仍然是 null。

我需要将其转换为同步方法（阻止并返回结果），如下所示：

根据您的要求，您需要阻塞当前线程，直到异步任务完成。为此，您可以使用Semaphore 或CountDownLatch。例如，使用Semaphore，您的方法可能如下所示：

fun mySyncMethod(params:Type) {
    Log.d("tag", "1")
    val result = null
    val semaphore = Semaphore(0)
    myAsyncMethod(params) {
        Log.d("tag", "3")
        result = it
        semaphore.release()
    }
    Log.d("tag", "2")
    semaphore.acquireUninterruptibly()
    Log.d("tag", "4")
    return result
}