使用 Kotlin 协程创建动态代理

Posted 2022-12-06 码农乐园

够在运行时实现接口，并决定如何在调用方法时动态执行方法。这对于在装饰器模式中描述的现有实现（尤其是来自第三方库）周围添加附加功能非常有用。然而，Kotlin协程在创建动态代理时引入了一系列新问题。我们将探讨如何正确检测和动态调用在接口上声明的挂起函数。

设置

首先，使用 Gradle 创建一个简单的 Kotlin 控制台应用程序并添加以下依赖项。

implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.4.3")

为简单起见，我们将把所有内容都放在 main.kt 文件中，从ITest我们将代理的接口开始。请注意，该接口混合了常规和挂起功能，其中一些会引发异常。即使在 Kotlin 中没有检查异常，我们也需要标记抛出的方法，以便 JVM在我们的调用处理程序传播异常时不会产生UndeclaredThrowableException 。

interface ITest 
    fun test(): String?
    @Throws(Exception::class)
    fun testException(): String?
    suspend fun testSuspend(): String?
    @Throws(Exception::class)
    suspend fun testExceptionSuspend(): String?

ITest我们将通过上述接口的简单底层实现来跟进。

class TestImpl: ITest 
    override fun test(): String? 
        return "test result"
    

    @Throws(Exception::class)
    override fun testException(): String? 
        throw Exception("You called testException().")
    

    override suspend fun testSuspend(): String? 
        delay(1000)
        return "testSuspend result"
    

    @Throws(Exception::class)
    override suspend fun testExceptionSuspend(): String? 
        delay(1000)
        throw Exception("You called testExceptionSuspend().")
    

目标

假设我们想用一些异常处理来装饰 TestImpl 对象。出于我们的目的，我们将异常处理定义为捕获并打印出任何异常，然后将其包装并重新抛出为我们自己的WrappedTestException. 所以让我们在 main.kt 文件中定义它。

class WrappedTestException(cause: Throwable): Exception(cause)

缺乏经验的方法

如果你习惯用 Java 做这类事情，很容易忘记协程。在 Kotlin 中创建动态代理时，不考虑挂起函数同样容易。让我们看看如果我们天真地这样做会发生什么。我们将在 main.kt 中创建下面的通用 InvocationHandler。此类将调用委托给传入的任何通用实例。它捕获任何InvocationTargetException，打印我们的委托实例抛出的底层异常，然后重新抛出它包装在我们的WrappedTestException.

/**
 * File: main.kt
 *
 * Dynamic proxy that naively catches and prints out exceptions and re-throws them wrapped as WrappedTestExceptions.
 */
class NaiveExceptionLogger<T>(private val instance: T): InvocationHandler 
    override fun invoke(proxy: Any?, method: Method?, args: Array<out Any>?): Any? 
        try 
            val nonNullArgs = args ?: arrayOf()
            return method?.invoke(instance, *nonNullArgs)
         catch(e: InvocationTargetException) 
            e.targetException?.let targetException ->
                println("Naively caught underlying target exception $targetException")
                throw WrappedTestException(targetException)
             ?: throw WrappedTestException(e)
        
    

现在，让我们使用以下方法创建并运行TestFactory它main。我们需要在内部执行，因为我们正在从同步方法runBlocking中触发挂起函数。main你认为会发生什么？（剧透：它会失败，但是如何以及为什么？）

object TestFactory 
    fun createNaively(): ITest 
        return Proxy.newProxyInstance(
            ITest::class.java.classLoader,
            arrayOf<Class<*>>(ITest::class.java),
            NaiveExceptionLogger(TestImpl())) as ITest
    



fun main(args: Array<String>) 
    runBlocking 
        val test = TestFactory.createNaively()
        println(test.test())
        println(test.testSuspend())
        try 
            test.testException()
            throw IllegalStateException("Did not catch testException()")
         catch(e: WrappedTestException)  
        try 
            test.testExceptionSuspend()
            throw IllegalStateException("Did not catch testExceptionSuspend()")
         catch(e: WrappedTestException)  
    

当你运行它时，你应该得到以下输出：

test result
testSuspend result
Naively caught underlying target exception java.lang.Exception: You called testException().
Exception in thread "main" java.lang.Exception: You called testExceptionSuspend().
    at TestImpl.testExceptionSuspend(main.kt:39)
    at TestImpl$testExceptionSuspend$1.invokeSuspend(main.kt)
    at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
    at kotlinx.coroutines.DispatchedTaskKt.resume(DispatchedTask.kt:234)
    at kotlinx.coroutines.DispatchedTaskKt.dispatch(DispatchedTask.kt:166)
    at kotlinx.coroutines.CancellableContinuationImpl.dispatchResume(CancellableContinuationImpl.kt:369)
    at kotlinx.coroutines.CancellableContinuationImpl.resumeImpl(CancellableContinuationImpl.kt:403)
    at kotlinx.coroutines.CancellableContinuationImpl.resumeImpl$default(CancellableContinuationImpl.kt:395)
    at kotlinx.coroutines.CancellableContinuationImpl.resumeUndispatched(CancellableContinuationImpl.kt:491)
    at kotlinx.coroutines.EventLoopImplBase$DelayedResumeTask.run(EventLoop.common.kt:489)
    at kotlinx.coroutines.EventLoopImplBase.processNextEvent(EventLoop.common.kt:274)
    at kotlinx.coroutines.BlockingCoroutine.joinBlocking(Builders.kt:84)
    at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking(Builders.kt:59)
    at kotlinx.coroutines.BuildersKt.runBlocking(Unknown Source)
    at kotlinx.coroutines.BuildersKt__BuildersKt.runBlocking$default(Builders.kt:38)
    at kotlinx.coroutines.BuildersKt.runBlocking$default(Unknown Source)
    at MainKt.main(main.kt:126)

Process finished with exit code 1

查看堆栈跟踪，很明显该testExceptionSuspend方法抛出的异常没有被捕获、记录和包装。为什么，你问？好吧，因为从技术上讲，该方法不会“抛出”传统意义上的异常。由于它是一个挂起函数，它实际上返回一个Continuation，它稍后会以结果或异常恢复。将 Continuation 视为类似于 javascript 中的 Promise 会有所帮助。NaiveExceptionLogger一旦返回 Continuation，我们的画面就会消失，早在它以实际结果恢复之前。从它的角度来看，调用成功完成，没有异常，并返回了一个值；故事结局。

正确的方法

如果我们想正确的从testExceptionSuspend方法中捕获异常，除了拦截方法调用外，还必须在Continuation恢复时拦截结果。要做到这一点，我们必须掌握这个延续，但它从何而来？Kotlin 在所有标记为挂起的函数的方法签名中的“秘密”最后一个参数中编译它们。除了最终结果之外，这些 Continuation 对象还带有CoroutineContext。因此，这里的基本方法是检查被调用方法的最后一个参数。如果它为 null 或不是 Continuation，那么这是一个我们可以调用的常规函数，就像我们在上面的简单示例中所做的那样。但是，如果我们找到一个 Continuation 作为最后一个 arg，我们需要执行以下操作：

1. 包装此基础 Continuation 并覆盖该resumeWith方法。

2. 如果我们的包装器 Continuation 因异常而恢复，请记录它并使用我们的 custom 恢复底层的 Continuation WrappedTestException。否则，以常规结果恢复底层证券。

3. 接下来，我们需要在CoroutineScope中执行这个方法。不要试图在GlobalScope中启动此调用。这样做会破坏 Kotlin 中的结构化并发模型，并且调用可能不会像预期的那样被清理或取消。CoroutineScope幸运的是，我们可以使用在我们底层 Continuation 中找到的上下文来正确构建一个。

我们也可以根据自己的需求来组合这个上下文，例如通过 Dispatchers启动在IO 线程池中运行的调用。调用该方法时，我们现在只需要更改与我们的新的和改进的 Continuation 一起提供的最后一个参数。

既然我们已经在 IO 线程池中启动了方法调用，那么我们从动态调用的方法中返回什么？Kotlin 对挂起函数有一个神奇的返回值，它告诉运行时挂起执行并允许当前线程继续执行其他工作。它是 kotlin.coroutines.intrinsics。COROUTINE_SUSPENDED。 

这是所有东西放在一起时的样子：

/**
 * File: main.kt
 *
 * Dynamic proxy that correctly catches and prints out exceptions with proper handling for coroutines.
 * Rethrows caught exceptions as WrappedTestException
 */
class CorrectExceptionLogger<T>(private val instance: T): InvocationHandler 
    override fun invoke(proxy: Any?, method: Method?, args: Array<out Any>?): Any? 
        val nonNullArgs = args ?: arrayOf()
        try 
            val lastArg = nonNullArgs.lastOrNull()
            if(lastArg == null || lastArg !is Continuation<*>) 
                // not a suspend func, just invoke regularly
                return method?.invoke(instance, *nonNullArgs)
             else 
                // Step 1: Wrap the underlying continuation to intercept exceptions.
                @Suppress("UNCHECKED_CAST")
                val originalContinuation = lastArg as Continuation<Any?>
                val wrappedContinuation = object: Continuation<Any?> 
                    override val context: CoroutineContext get() = originalContinuation.context
                    override fun resumeWith(result: Result<Any?>) 
                        result.exceptionOrNull()?.let err ->
                            // Step 2: log intercepted exception and resume with our custom wrapped exception.
                            println("Correctly caught underlying coroutine exception $err")
                            originalContinuation.resumeWithException(WrappedTestException(err))
                         ?: originalContinuation.resumeWith(result)
                    
                
                // Step 3: launch the suspend function with our wrapped continuation using the underlying scope and context, but force it to run in the IO thread pool
                CoroutineScope(originalContinuation.context).launch(Dispatchers.IO + originalContinuation.context) 
                    val argumentsWithoutContinuation = nonNullArgs.take(nonNullArgs.size - 1)
                    val newArgs = argumentsWithoutContinuation + wrappedContinuation
                    method?.invoke(instance, *newArgs.toTypedArray())
                
                return kotlin.coroutines.intrinsics.COROUTINE_SUSPENDED
            
         catch(e: InvocationTargetException) 
            e.targetException?.let targetException ->
                println("Correctly caught underlying exception $targetException")
                throw WrappedTestException(targetException)
             ?: throw WrappedTestException(e)
        
    

运行！

好的，如果您还在这里，让我们将这个新实现添加到我们的TestFactory并运行它。

object TestFactory 
    fun createNaively(): ITest 
        return Proxy.newProxyInstance(
            ITest::class.java.classLoader,
            arrayOf<Class<*>>(ITest::class.java),
            NaiveExceptionLogger(TestImpl())) as ITest
    

    fun createCorrectly(): ITest 
        return Proxy.newProxyInstance(
            ITest::class.java.classLoader,
            arrayOf<Class<*>>(ITest::class.java),
            CorrectExceptionLogger(TestImpl())) as ITest
    



fun main(args: Array<String>) 
    runBlocking 
        val test = TestFactory.createCorrectly()
        println(test.test())
        println(test.testSuspend())
        try 
            test.testException()
            throw IllegalStateException("Did not catch testException()")
         catch(e: WrappedTestException)  
        try 
            test.testExceptionSuspend()
            throw IllegalStateException("Did not catch testExceptionSuspend()")
         catch(e: WrappedTestException)  
    

您的输出应如下所示。如果是这样——恭喜！– 您现在已经创建了一个通用的动态代理，它可以与常规和挂起函数一起使用。

test result
testSuspend result
Correctly caught underlying exception java.lang.Exception: You called testException().
Correctly caught underlying coroutine exception java.lang.Exception: You called testExceptionSuspend().

Process finished with exit code 0