Kotlin 语言扩展

你当前正在浏览历史版本 Vert.x 4.4.9 的文档。访问本页面的最新版本。

vertx-lang-kotlin-coroutines

vertx-lang-kotlin-coroutines 集成了 Kotlin Coroutine，用于执行异步操作和处理事件。这带来了一种看似顺序代码的编程模型，同时又不会阻塞内核线程。

简介

与许多传统应用平台相比，Vert.x 的核心优势之一在于其（对内核线程）几乎完全是非阻塞的。这使得基于 Vert.x 的应用程序能够仅使用极少量的内核线程来处理海量并发（例如大量的连接和消息），从而释放极佳的可伸缩性。

Vert.x 的非阻塞特性催生了异步 API。异步 API 可以表现为多种形式，包括回调（Callback）、Promise、纤程（Fiber）或响应式扩展（Reactive Extensions）。 Vert.x 的核心 API 采用了回调风格，但同时也支持诸如 RxJava 等其它模型。

在某些情况下，使用异步 API 进行编程可能会比使用传统/顺序代码风格更具挑战性，尤其是在需要按顺序执行多个操作时。此外，在使用异步 API 时，错误传播通常也更为复杂。

vertx-lang-kotlin-coroutines 采用了 Coroutine。Coroutine 是一种极轻量级的线程，并不与底层内核线程直接对应，因此当某个 Coroutine 需要“阻塞”时，它会被挂起（suspend）并释放其当前的内核线程，以便其它 Coroutine 可以继续处理事件。

vertx-lang-kotlin-coroutines 使用 kotlinx.coroutines 来实现 Coroutine。

在 Vert.x 上下文中运行 Coroutine

导入 io.vertx.kotlin.coroutines.VertxCoroutine 后，即可通过 GlobalScope.launch 方法将代码块作为 Coroutine 运行在全局（Global）应用作用域中（生命周期与应用程序绑定）：

val vertx = Vertx.vertx()

GlobalScope.launch(vertx.dispatcher()) {
  val timerId = awaitEvent<Long> { handler ->
    vertx.setTimer(1000, handler)
  }
  println("Event fired from timer with id $timerId")
}

vertx.dispatcher() 返回一个 Coroutine Dispatcher，它使用 Vert.x EventLoop 来执行 Coroutine。

awaitEvent 函数会挂起 Coroutine 的执行，直到定时器触发，然后使用传递给 Handler 的值恢复 Coroutine。

关于 Handler、事件以及事件流的更多细节，将在接下来的各节中详细介绍。

你可以将代码作为 io.vertx.kotlin.coroutines.CoroutineVerticle 的实例进行部署，这是专门为 Kotlin Coroutine 定制的 Verticle 类型。CoroutineVerticle 类实现了 kotlinx.coroutines.CoroutineScope 接口，使得所有 Coroutine 构建器方法默认绑定到该 Verticle 上下文。你应该重写 Verticle 的挂起函数 start() 以及（可选的）挂起函数 stop()：

class MyVerticle : CoroutineVerticle() {
  override suspend fun start() {
    // ...
  }

  override suspend fun stop() {
    // ...
  }
}

以下所有代码示例均假定在 CoroutineVerticle 的实现内部运行，但你也可以将所有 <builder> { .. } 调用替换为 GlobalScope.<builder> { .. }，以改用应用全局作用域。

获取单次异步结果

Vert.x 中的许多异步操作都以 Handler<AsyncResult<T>> 作为最后一个参数。例如使用 Vert.x Mongo 客户端执行对象检索，或者发送 Event Bus 消息并等待应答。

这可以通过使用 awaitResult 方法来实现，该方法会返回结果值或抛出异常。

Coroutine 会被挂起，直到事件被处理完毕，期间不会阻塞任何内核线程。

该方法的调用方式为：以代码块的形式指定需要执行的异步操作，并在运行时传递给 Handler。

示例如下：

suspend fun awaitResultExample() {
  val consumer = vertx.eventBus().localConsumer<String>("a.b.c")
  consumer.handler { message ->
    println("Consumer received: ${message.body()}")
    message.reply("pong")
  }

  // Send a message and wait for a reply
  val reply = awaitResult<Message<String>> { h ->
    vertx.eventBus().request("a.b.c", "ping", h)
  }
  println("Reply received: ${reply.body()}")
}

当该代码块发生失败时，调用方可以使用常规的异常 try/catch 结构将其作为异常进行处理：

suspend fun awaitResultFailureExample() {
  val consumer = vertx.eventBus().localConsumer<String>("a.b.c")
  consumer.handler { message ->
    // The consumer will get a failure
    message.fail(0, "it failed!!!")
  }

  // Send a message and wait for a reply
  try {
    awaitResult<Message<String>> { h ->
      vertx.eventBus().request("a.b.c", "ping", h)
    }
  } catch (e: ReplyException) {
    // Handle specific reply exception here
    println("Reply failure: ${e.message}")
  }
}

获取单次事件

处理单次事件（而不处理后续发生的事件，如果有的话）可以通过 awaitEvent 函数实现：

suspend fun awaitEventExample() {
  val id = awaitEvent<Long> { h -> vertx.setTimer(2000L, h) }
  println("This should be fired in 2s by some time with id=$id")
}

获取单次 Worker 结果

处理阻塞式计算可以通过 awaitBlocking 函数实现：

suspend fun awaitBlockingExample() {
  awaitBlocking {
    Thread.sleep(1000)
    "some-string"
  }
}

事件流

在 Vert.x API 的许多地方，事件流都是通过 Handler 进行处理的。例如 Event Bus 消息消费者和 HTTP 服务器请求。

ReceiveChannelHandler 类允许通过（可挂起的）receive 方法来接收事件：

suspend fun streamExample() {
  val adapter = vertx.receiveChannelHandler<Message<Int>>()
  vertx.eventBus().localConsumer<Int>("a.b.c").handler(adapter)

  // Send 15 messages
  for (i in 0..15) vertx.eventBus().send("a.b.c", i)

  // Receive the first 10 messages
  for (i in 0..10) {
    val message = adapter.receive()
    println("Received: ${message.body()}")
  }
}

等待 Vert.x 异步结果完成

Vert.x 4 提供了 Future，且 Future 拥有一个用于等待异步结果的 coAwait() 挂起方法。

针对 Vert.x Future 实例的 coAwait 扩展方法会挂起 Coroutine 直至其完成；完成后，该方法将返回对应 AsyncResult<T> 对象的值或抛出其失败异常。

suspend fun awaitingFuture(anotherFuture: Future<String>) {
  // Getting a future
  val httpServerFuture = vertx.createHttpServer()
    .requestHandler { req -> req.response().end("Hello!") }
    .listen(8000)

  val httpServer = httpServerFuture.await()
  println("HTTP server port: ${httpServer.actualPort()}")

  // It also works for composite futures
  val result = CompositeFuture.all(httpServerFuture, anotherFuture).await()
  if (result.succeeded()) {
    println("The server is now running!")
  } else {
    result.cause().printStackTrace()
  }
}

Channel

Channel 类似于 Java 的 BlockingQueue，不同之处在于它们会挂起 Coroutine 而不是阻塞线程：

向已满的 Channel 发送值会挂起 Coroutine
从空的 Channel 接收值也会挂起 Coroutine

通过 toChannel 扩展方法，可以将 Vert.x 的 ReadStream 和 WriteStream 适配为 Channel。

这些适配器负责管理背压（Back-pressure）以及流的终止：

ReadStream<T> 适配为 ReceiveChannel<T>
WriteStream<T> 适配为 SendChannel<T>

接收数据

当你需要处理一系列相关联的值构成的流时，Channel 会非常有用：

suspend fun handleTemperatureStream() {
  val stream = vertx.eventBus().consumer<Double>("temperature")
  val channel = stream.toChannel(vertx)

  var min = Double.MAX_VALUE
  var max = Double.MIN_VALUE

  // Iterate until the stream is closed
  // Non-blocking
  for (msg in channel) {
    val temperature = msg.body()
    min = Math.min(min, temperature)
    max = Math.max(max, temperature)
  }

  // The stream is now closed
}

它在解析协议时也非常有用。我们将构建一个非阻塞的 HTTP 请求解析器，以展示 Channel 的威力。

我们将依赖 RecordParser 将 Buffer 流按 \r\n 分隔切片为 Buffer 流。

以下是解析器的初始版本，仅处理 HTTP 请求行（Request-line）：

vertx.createNetServer().connectHandler { socket ->

  // The record parser provides a stream of buffers delimited by \r\n
  val stream = RecordParser.newDelimited("\r\n", socket)

  // Convert the stream to a Kotlin channel
  val channel = stream.toChannel(vertx)

  // Run the coroutine
  launch {

    // Receive the request-line
    // Non-blocking
    val line = channel.receive().toString().split(" ")
    val method = line[0]
    val uri = line[1]

    println("Received HTTP request ($method, $uri)")

    // Still need to parse headers and body...
  }
}

解析请求行就像在 Channel 上调用 receive 一样简单。

下一步是通过持续接收数据块直到遇到空数据块来解析 HTTP 请求头（Headers）：

// Receive HTTP headers
val headers = HashMap<String, String>()
while (true) {

  // Non-blocking
  val header = channel.receive().toString()

  // Done with parsing headers
  if (header.isEmpty()) {
    break
  }

  val pos = header.indexOf(':')
  headers[header.substring(0, pos).toLowerCase()] = header.substring(pos + 1).trim()
}

println("Received HTTP request ($method, $uri) with headers ${headers.keys}")

最后，我们通过处理可选的请求体（Request Body）来完成该解析器：

// Receive the request body
val transferEncoding = headers["transfer-encoding"]
val contentLength = headers["content-length"]

val body: Buffer?
if (transferEncoding == "chunked") {

  // Handle chunked encoding, e.g
  // 5\r\n
  // HELLO\r\n
  // 0\r\n
  // \r\n

  body = Buffer.buffer()
  while (true) {

    // Parse length chunk
    // Non-blocking
    val len = channel.receive().toString().toInt(16)
    if (len == 0) {
      break
    }

    // The stream is flipped to parse a chunk of the exact size
    stream.fixedSizeMode(len + 2)

    // Receive the chunk and append it
    // Non-blocking
    val chunk = channel.receive()
    body.appendBuffer(chunk, 0, chunk.length() - 2)

    // The stream is flipped back to the \r\n delimiter to parse the next chunk
    stream.delimitedMode("\r\n")
  }
} else if (contentLength != null) {

  // The stream is flipped to parse a body of the exact size
  stream.fixedSizeMode(contentLength.toInt())

  // Non-blocking
  body = channel.receive()
} else {
  body = null
}

val bodySize = body?.length() ?: 0
println("Received HTTP request ($method, $uri) with headers ${headers.keys} and body with size $bodySize")

发送数据

使用 Channel 发送数据非常直接：

suspend fun sendChannel(httpResponse : HttpServerResponse) {
  val channel = httpResponse.toChannel(vertx)

  while (true) {
    val buffer = readBuffer()

    // Broadcast the temperature
    // Non-blocking but could be suspended
    channel.send(buffer)

    // Wait for one second
    awaitEvent<Long> { vertx.setTimer(1000, it) }
  }
}

SendChannel#send 与 WriteStream#write 均为非阻塞操作，不同之处在于当 Channel 满时 SendChannel#send 会挂起执行。

延迟、取消与超时

Vert.x Dispatcher 通过 Vert.x 定时器完全支持 Coroutine 的 delay 函数：

launch {
  // Set a one second Vertx timer
  delay(1000)
}

定时器支持取消：

val job = launch {
  // Set a one second Vertx timer
  while (true) {
    delay(1000)
    // Do something periodically
  }
}

// Sometimes later
job.cancel()

取消是协作式的（Cooperative）

你还可以使用 withTimeout 函数来设置超时调度：

launch {
  try {
    val id = withTimeout<String>(1000) {
      awaitEvent { anAsyncMethod(it) }
    }
  } catch (e: TimeoutCancellationException) {
    // Cancelled
  }
}

Coroutine 构建器

只要有可用的 CoroutineScope 实例，Vert.x 就可以与所有 Coroutine 构建器协同工作：launch、async、produce 等。有几点重要事项需要牢记：

runBlocking 不需要 CoroutineScope，并且绝不能在 Vert.x EventLoop 线程中调用。
为避免内存泄漏，请始终使用 coroutineScope {..} 来定义子作用域。通过这种方式，如果作用域内的某个 Coroutine 发生失败，该作用域内定义的所有其它 Coroutine 也都将被一并取消。

Vert.x 还提供了一个返回 io.vertx.core.Future 实例的 Coroutine 构建器：


// Can be called on any thread
val future1: Future<String> = vertxFuture(vertx) {
  computeSomethingWithSuspendingFunction()
}

// Can be called only when running on a Vert.x context
val future2: Future<String> = vertxFuture {
  computeSomethingWithSuspendingFunction()
}

Coroutine 互操作性

Vert.x 的集成经过专门设计，能够与 Kotlin Coroutine 完全互操作：

在使用 Vert.x Dispatcher 时，kotlinx.coroutines.sync.Mutex 会在 EventLoop 线程上执行

Vert.x for Kotlin

Vert.x for Groovy