Kotlin 实现多线程请求对冲？答案

【问题标题】：Kotlin to achieve multithread request hedging?Kotlin 实现多线程请求对冲？
【发布时间】：2019-11-06 18:26:50
【问题描述】：

Spring 的 reactor 有一个有趣的功能：Hedging。这意味着产生许多请求并获得第一个返回的结果，并自动清理其他上下文。 Josh Long 最近一直在积极推广这个功能。谷歌搜索Spring reactor hedging 显示相对结果。如果有人好奇，here 是示例代码。简而言之，Flux.first() 简化了所有底层的麻烦，令人印象深刻。

我想知道如何使用 Kotlin 的 coroutine 和 multithread 来实现这一点（也许使用 Flow 或 Channel ）。我想到了一个简单的场景：一个服务接受 longUrl 并将 longUrl 生成到许多 URL 缩短服务（例如 IsGd 、 TinyUrl ...），并返回第一个返回的 URL ...（并终止/清理其他线程/协程资源)

有一个接口UrlShorter 定义了这项工作：

interface UrlShorter {
  fun getShortUrl(longUrl: String): String?
}

并且有三种实现，一种用于is.gd，另一种用于tinyUrl，第三种是阻塞10秒并返回null的Dumb实现：

class IsgdImpl : UrlShorter {
  override fun getShortUrl(longUrl: String): String? {
    logger.info("running : {}", Thread.currentThread().name)
    // isGd api url blocked by SO , it sucks . see the underlaying gist for full code
    val url = "https://is.gd/_create.php?format=simple&url=%s".format(URLEncoder.encode(longUrl, "UTF-8"))
    return Request.Get(url).execute().returnContent().asString().also {
      logger.info("returning {}", it)
    }
  }
}

class TinyImpl : UrlShorter {
  override fun getShortUrl(longUrl: String): String? {
    logger.info("running : {}", Thread.currentThread().name)
    val url = "http://tinyurl.com/_api-create.php?url=$longUrl" // sorry the URL is blocked by stackoverflow , see the underlaying gist for full code
    return Request.Get(url).execute().returnContent().asString().also {
      logger.info("returning {}", it)
    }
  }
}

class DumbImpl : UrlShorter {
  override fun getShortUrl(longUrl: String): String? {
    logger.info("running : {}", Thread.currentThread().name)
    TimeUnit.SECONDS.sleep(10)
    return null
  }
}

还有一个UrlShorterService 接受所有UrlShorter 实现，并尝试生成协程并获得第一个结果。

这是我的想法：

@ExperimentalCoroutinesApi
@FlowPreview
class UrlShorterService(private val impls: List<UrlShorter>) {


  private val es: ExecutorService = Executors.newFixedThreadPool(impls.size)
  private val esDispatcher = es.asCoroutineDispatcher()

  suspend fun getShortUrl(longUrl: String): String {
    return method1(longUrl) // there are other methods , with different ways...
  }

  private inline fun <T, R : Any> Iterable<T>.firstNotNullResult(transform: (T) -> R?): R? {
    for (element in this) {
      val result = transform(element)
      if (result != null) return result
    }
    return null
  }

客户端也很简单：

@ExperimentalCoroutinesApi
@FlowPreview
class UrlShorterServiceTest {

  @Test
  fun testHedging() {
    val impls = listOf(DumbImpl(), IsgdImpl(), TinyImpl()) // Dumb first
    val service = UrlShorterService(impls)
    runBlocking {
      service.getShortUrl("https://www.google.com").also {
        logger.info("result = {}", it)
      }
    }
  }
}

请注意，我将DumbImpl 放在首位，因为我希望它可能首先生成并阻塞在其线程中。其他两种实现都可以得到结果。

好的，问题来了，在kotlin中如何实现对冲？我尝试以下方法：

  private suspend fun method1(longUrl: String): String {
    return impls.asSequence().asFlow().flatMapMerge(impls.size) { impl ->
      flow {
        impl.getShortUrl(longUrl)?.also {
          emit(it)
        }
      }.flowOn(esDispatcher)
    }.first()
      .also { esDispatcher.cancelChildren() } // doesn't impact the result
  }

我希望method1 应该可以工作，但它总共执行 10 秒：

00:56:09,253 INFO  TinyImpl - running : pool-1-thread-3
00:56:09,254 INFO  DumbImpl - running : pool-1-thread-1
00:56:09,253 INFO  IsgdImpl - running : pool-1-thread-2
00:56:11,150 INFO  TinyImpl - returning // tiny url blocked by SO , it sucks
00:56:13,604 INFO  IsgdImpl - returning // idGd url blocked by SO , it sucks
00:56:19,261 INFO  UrlShorterServiceTest$testHedging$1 - result = // tiny url blocked by SO , it sucks

然后，我认为其他方法 2 ，方法 3 ，方法 4 ，方法 5 ......但都不起作用：

  /**
   * 00:54:29,035 INFO  IsgdImpl - running : pool-1-thread-3
   * 00:54:29,036 INFO  DumbImpl - running : pool-1-thread-2
   * 00:54:29,035 INFO  TinyImpl - running : pool-1-thread-1
   * 00:54:30,228 INFO  TinyImpl - returning // tiny url blocked by SO , it sucks
   * 00:54:30,797 INFO  IsgdImpl - returning // idGd url blocked by SO , it sucks
   * 00:54:39,046 INFO  UrlShorterServiceTest$testHedging$1 - result = // idGd url blocked by SO , it sucks
   */
  private suspend fun method2(longUrl: String): String {
    return withContext(esDispatcher) {
      impls.map { impl ->
        async(esDispatcher) {
          impl.getShortUrl(longUrl)
        }
      }.firstNotNullResult { it.await() } ?: longUrl
    }
  }

  /**
   * 00:52:30,681 INFO  IsgdImpl - running : pool-1-thread-2
   * 00:52:30,682 INFO  DumbImpl - running : pool-1-thread-1
   * 00:52:30,681 INFO  TinyImpl - running : pool-1-thread-3
   * 00:52:31,838 INFO  TinyImpl - returning // tiny url blocked by SO , it sucks
   * 00:52:33,721 INFO  IsgdImpl - returning // idGd url blocked by SO , it sucks
   * 00:52:40,691 INFO  UrlShorterServiceTest$testHedging$1 - result = // idGd url blocked by SO , it sucks
   */
  private suspend fun method3(longUrl: String): String {
    return coroutineScope {
      impls.map { impl ->
        async(esDispatcher) {
          impl.getShortUrl(longUrl)
        }
      }.firstNotNullResult { it.await() } ?: longUrl
    }
  }

  /**
   * 01:58:56,930 INFO  TinyImpl - running : pool-1-thread-1
   * 01:58:56,933 INFO  DumbImpl - running : pool-1-thread-2
   * 01:58:56,930 INFO  IsgdImpl - running : pool-1-thread-3
   * 01:58:58,411 INFO  TinyImpl - returning // tiny url blocked by SO , it sucks
   * 01:58:59,026 INFO  IsgdImpl - returning // idGd url blocked by SO , it sucks
   * 01:59:06,942 INFO  UrlShorterServiceTest$testHedging$1 - result =  // idGd url blocked by SO , it sucks
   */
  private suspend fun method4(longUrl: String): String {
    return withContext(esDispatcher) {
      impls.map { impl ->
        async {
          impl.getShortUrl(longUrl)
        }
      }.firstNotNullResult { it.await() } ?: longUrl
    }
  }

我不熟悉Channel，异常见谅↓

  /**
   * 01:29:44,460 INFO  UrlShorterService$method5$2 - channel closed
   * 01:29:44,461 INFO  DumbImpl - running : pool-1-thread-2
   * 01:29:44,460 INFO  IsgdImpl - running : pool-1-thread-3
   * 01:29:44,466 INFO  TinyImpl - running : pool-1-thread-1
   * 01:29:45,765 INFO  TinyImpl - returning // tiny url blocked by SO , it sucks
   * 01:29:46,339 INFO  IsgdImpl - returning // idGd url blocked by SO , it sucks
   *
   * kotlinx.coroutines.channels.ClosedSendChannelException: Channel was closed
   *
   */
  private suspend fun method5(longUrl: String): String {
    val channel = Channel<String>()

    withContext(esDispatcher) {
      impls.forEach { impl ->
        launch {
          impl.getShortUrl(longUrl)?.also {
            channel.send(it)
          }
        }
      }
      channel.close()
      logger.info("channel closed")
    }

    return channel.consumeAsFlow().first()
  }

好的，我不知道有没有其他方法...但是以上都不起作用...所有阻塞至少10秒（被DumbImpl阻塞）。

完整的源代码可以在github gist 上找到。

如何在 kotlin 中实现对冲？通过Deferred 或Flow 或Channel 或任何其他更好的想法？谢谢。

提交问题后，我发现所有 tinyurl 、isGd url 都被 SO 屏蔽了。真的很烂！

【问题讨论】：

您专门创建了一个在整个持续时间内不可取消的获取方法。那是真正的方法会做的吗？如果是这样，那么您必须泄漏它们并且不强制执行结构化并发规则。让他们在后台慢慢来。但我真的不知道这是否是一个聪明的设计。
那么，在您看来，interface UrlShorter { fun getShortUrl(longUrl: String): String? } 有问题吗？我最初的想法是，它只是一个阻塞方法，每个实现不需要知道它会被阻塞或在挂起函数中运行调用。如果在 (timeout or API invoking ...) 内部发生错误，它只会捕获 Exception 并返回 null（为简洁起见发出代码）。如果设计不好，如何重新设计呢？谢谢。
如果您正在使用阻塞调用，那么协程无论如何都不会提供太多价值。你可以用老式的 Java 执行器解决同样的问题。但是，如果您确保不等待协程范围完成，您仍然可以使用协程来完成。
那么，如果 UrlShorter 定义了 suspend fun getShortUrl(longUrl : String) : String? ，那么如何重新设计这样的架构，而不出现泄漏？如果您提供一个简单的代码，我将不胜感激。
只是让它suspend fun 没有帮助，你必须在其中进行暂停调用。如果您可以使用（如果您只是发出网络请求），那么可以正确解决。在这种情况下，您的解决方案之一也可能会开始工作。我也开始草拟一个解决方案，如果我从中得到任何好处，我会写一个答案。

标签： multithreading kotlin reactive-programming project-reactor kotlin-coroutines

【解决方案1】：

如果您要并行执行的实际工作包括网络提取，则应选择异步网络库，以便可以正确使用非阻塞协同程序。例如，从版本 11 开始，JDK 提供了一个异步 HTTP 客户端，您可以按如下方式使用它：

val httpClient: HttpClient = HttpClient.newHttpClient()

suspend fun httpGet(url: String): String = httpClient
        .sendAsync(
                HttpRequest.newBuilder().uri(URI.create(url)).build(),
                BodyHandlers.ofString())
        .await()
        .body()

这是一个在给定上述可暂停实现的情况下完成请求对冲的函数：

class UrlShortenerService(
        private val impls: List<UrlShortener>
) {
    suspend fun getShortUrl(longUrl: String): String? = impls
            .asFlow()
            .flatMapMerge(impls.size) { impl ->
                flow<String?> {
                    try {
                        impl.getShortUrl(longUrl)?.also { emit(it) }
                    }
                    catch (e: Exception) { 
                        // maybe log it, but don't let it propagate
                    }
                }
            }
            .onCompletion { emit(null) }
            .first()
}

请注意没有任何自定义调度程序，您不需要它们来进行可暂停的工作。任何调度程序都可以，所有工作都可以在一个线程中运行。

当您的所有 URL 缩短器都失败时，onCompletion 部分会开始行动。在这种情况下，flatMapMerge 阶段不会发出任何内容，first() 会在没有额外的null 注入流的情况下陷入僵局。

为了测试它，我使用了以下代码：

class Shortener(
        private val delay: Long
) : UrlShortener {
    override suspend fun getShortUrl(longUrl: String): String? {
        delay(delay * 1000)
        println("Shortener $delay completing")
        if (delay == 1L) {
            throw Exception("failed service")
        }
        if (delay == 2L) {
            return null
        }
        return "shortened after $delay seconds"
    }
}

suspend fun main() {
    val shorteners = listOf(
            Shortener(4),
            Shortener(3),
            Shortener(2),
            Shortener(1)
    )
    measureTimeMillis {
        UrlShortenerService(shorteners).getShortUrl("bla").also {
            println(it)
        }
    }.also {
        println("Took $it ms")
    }
}

这会练习各种失败情况，例如返回 null 或失败并出现异常。对于这段代码，我得到以下输出：

Shortener 1 completing
Shortener 2 completing
Shortener 3 completing
shortened after 3 seconds
Took 3080 ms

我们可以看到缩短器 1 和 2 已完成但失败，缩短器 3 返回了有效响应，并且缩短器 4 在完成之前被取消。我认为这符合要求。

如果您无法摆脱阻塞请求，您的实现将不得不启动num_impls * num_concurrent_requests 线程，这不是很好。但是，如果这是你能拥有的最好的，这里有一个实现，它可以对冲阻塞请求，但可以暂停和取消等待它们。它将向运行请求的工作线程发送中断信号，但如果您的库的 IO 代码是不可中断的，这些线程将挂起等待其请求完成或超时。

val es = Executors.newCachedThreadPool()

interface UrlShortener {
    fun getShortUrl(longUrl: String): String? // not suspendable!
}

class UrlShortenerService(
        private val impls: List<UrlShortener>
) {
    suspend fun getShortUrl(longUrl: String): String {
        val chan = Channel<String?>()
        val futures = impls.map { impl -> es.submit {
            try {
                impl.getShortUrl(longUrl)
            } catch (e: Exception) {
                null
            }.also { runBlocking { chan.send(it) } }
        } }
        try {
            (1..impls.size).forEach { _ ->
                chan.receive()?.also { return it }
            }
            throw Exception("All services failed")
        } finally {
            chan.close()
            futures.forEach { it.cancel(true) }
        }
    }
}

【讨论】：

嗨，您的 flatMapMerge 和 .onCompletion { emit(null) } 似乎不起作用。因为它似乎是第一个返回值......顺便说一句，我发现ktor 原生支持协程。示例代码在这里，仅供参考：github.com/smallufo/kotlinPlay/blob/master/hedging/src/test/…
这可能是实现中的一个错误，onCompletion 的文档建议仅使用这种用法将项目附加到上游流的末尾。
好吧，如果 .onCompletion { emit(null) } 按文档说明工作，那将是 IMO 的最佳解决方案。（等待 Jetbrains 修复错误）。
他们已经wrote a fix，我们可以期待下一个补丁版本。
好消息，fixing PR 现已合并。

【解决方案2】：

这基本上就是 select APi 的设计目的：

coroutineScope {
    select {
        impls.forEach { impl ->
            async {
               impl.getShortUrl(longUrl)
            }.onAwait { it }
        }
    }
    coroutineContext[Job].cancelChildren() // Cancel any requests that are still going.
}

请注意，这不会处理服务实现抛出的异常，如果您想实际处理这些异常，则需要使用带有自定义异常处理程序和过滤选择循环的 supervisorScope。

【讨论】：

您的代码似乎无法编译... impl.getShortUrl(longUrl) 返回 String? ，select 类型变为 select<String?> ，但方法应返回 String。和coroutineContext[Job].cancelChildren() 使块返回Unit ... Mmm .... 你能提供一个完整的方法吗？谢谢。
我玩过这种方法，但不值得，它的复杂性最终比通过单通道通信更糟糕。失败的处理是丑陋的，过滤选择循环是复杂的等等。