阅读(908) 赞(0)

Micronaut 响应式 HTTP 请求处理

2023-02-23 11:24:02 更新

如前所述，Micronaut 建立在 Netty 之上，Netty 是围绕事件循环模型和非阻塞 I/O 设计的。 Micronaut 在与请求线程（事件循环线程）相同的线程中执行在 @Controller bean 中定义的代码。

这使得在执行任何阻塞 I/O 操作（例如与 Hibernate/JPA 或 JDBC 的交互）时将这些任务卸载到不阻塞事件循环的单独线程池变得至关重要。

例如，以下配置将 I/O 线程池配置为具有 75 个线程的固定线程池（类似于 Tomcat 等传统阻塞服务器在每个请求线程模型中使用的线程池）：

配置IO线程池

Properties Yaml Toml Groovy Hocon JSON

micronaut.executors.io.type=fixed
micronaut.executors.io.nThreads=75

micronaut:
  executors:
    io:
      type: fixed
      nThreads: 75

[micronaut]
  [micronaut.executors]
    [micronaut.executors.io]
      type="fixed"
      nThreads=75

micronaut {
  executors {
    io {
      type = "fixed"
      nThreads = 75
    }
  }
}

{
  micronaut {
    executors {
      io {
        type = "fixed"
        nThreads = 75
      }
    }
  }
}

{
  "micronaut": {
    "executors": {
      "io": {
        "type": "fixed",
        "nThreads": 75
      }
    }
  }
}

要在 @Controller bean 中使用此线程池，您有多种选择。最简单的是使用@ExecuteOn 注释，它可以在类型或方法级别声明，以指示在哪个配置的线程池上运行控制器的方法：

使用@ExecuteOn

Java Groovy Kotlin

import io.micronaut.docs.http.server.reactive.PersonService;
import io.micronaut.docs.ioc.beans.Person;
import io.micronaut.http.annotation.Controller;
import io.micronaut.http.annotation.Get;
import io.micronaut.scheduling.TaskExecutors;
import io.micronaut.scheduling.annotation.ExecuteOn;

@Controller("/executeOn/people")
public class PersonController {

    private final PersonService personService;

    PersonController(PersonService personService) {
        this.personService = personService;
    }

    @Get("/{name}")
    @ExecuteOn(TaskExecutors.IO) // (1)
    Person byName(String name) {
        return personService.findByName(name);
    }
}

import io.micronaut.docs.http.server.reactive.PersonService
import io.micronaut.docs.ioc.beans.Person
import io.micronaut.http.annotation.Controller
import io.micronaut.http.annotation.Get
import io.micronaut.scheduling.TaskExecutors
import io.micronaut.scheduling.annotation.ExecuteOn

@Controller("/executeOn/people")
class PersonController {

    private final PersonService personService

    PersonController(PersonService personService) {
        this.personService = personService
    }

    @Get("/{name}")
    @ExecuteOn(TaskExecutors.IO) // (1)
    Person byName(String name) {
        personService.findByName(name)
    }
}

import io.micronaut.docs.http.server.reactive.PersonService
import io.micronaut.docs.ioc.beans.Person
import io.micronaut.http.annotation.Controller
import io.micronaut.http.annotation.Get
import io.micronaut.scheduling.TaskExecutors
import io.micronaut.scheduling.annotation.ExecuteOn

@Controller("/executeOn/people")
class PersonController (private val personService: PersonService) {

    @Get("/{name}")
    @ExecuteOn(TaskExecutors.IO) // (1)
    fun byName(name: String): Person {
        return personService.findByName(name)
    }
}

@ExecuteOn 注解用于在I/O线程池上执行操作

@ExecuteOn 注解的值可以是在 micronaut.executors 下定义的任何命名执行器。

一般来说，对于数据库操作，您需要配置一个与数据库连接池中指定的最大连接数相匹配的线程池。

@ExecuteOn 注释的替代方法是使用您选择的响应式库提供的工具。诸如 Project Reactor 或 RxJava 之类的响应式实现具有一个 subscribeOn 方法，该方法可让您更改执行用户代码的线程。例如：

响应式 subscribeOn 示例

Java Groovy Kotlin

import io.micronaut.docs.ioc.beans.Person;
import io.micronaut.http.annotation.Controller;
import io.micronaut.http.annotation.Get;
import io.micronaut.scheduling.TaskExecutors;
import jakarta.inject.Named;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import io.micronaut.core.async.annotation.SingleResult;
import java.util.concurrent.ExecutorService;

@Controller("/subscribeOn/people")
public class PersonController {

    private final Scheduler scheduler;
    private final PersonService personService;

    PersonController(
            @Named(TaskExecutors.IO) ExecutorService executorService, // (1)
            PersonService personService) {
        this.scheduler = Schedulers.fromExecutorService(executorService);
        this.personService = personService;
    }

    @Get("/{name}")
    @SingleResult
    Publisher<Person> byName(String name) {
        return Mono
                .fromCallable(() -> personService.findByName(name)) // (2)
                .subscribeOn(scheduler); // (3)
    }
}

import io.micronaut.docs.ioc.beans.Person
import io.micronaut.http.annotation.Controller
import io.micronaut.http.annotation.Get
import io.micronaut.scheduling.TaskExecutors
import jakarta.inject.Named
import reactor.core.publisher.Mono
import reactor.core.scheduler.Scheduler
import reactor.core.scheduler.Schedulers
import java.util.concurrent.ExecutorService

@Controller("/subscribeOn/people")
class PersonController {

    private final Scheduler scheduler
    private final PersonService personService

    PersonController(
            @Named(TaskExecutors.IO) ExecutorService executorService, // (1)
            PersonService personService) {
        this.scheduler = Schedulers.fromExecutorService(executorService)
        this.personService = personService
    }

    @Get("/{name}")
    Mono<Person> byName(String name) {
        return Mono
                .fromCallable({ -> personService.findByName(name) }) // (2)
                .subscribeOn(scheduler) // (3)
    }
}

import io.micronaut.docs.ioc.beans.Person
import io.micronaut.http.annotation.Controller
import io.micronaut.http.annotation.Get
import io.micronaut.scheduling.TaskExecutors
import java.util.concurrent.ExecutorService
import jakarta.inject.Named
import reactor.core.publisher.Mono
import reactor.core.scheduler.Scheduler
import reactor.core.scheduler.Schedulers


@Controller("/subscribeOn/people")
class PersonController internal constructor(
    @Named(TaskExecutors.IO) executorService: ExecutorService, // (1)
    private val personService: PersonService) {

    private val scheduler: Scheduler = Schedulers.fromExecutorService(executorService)

    @Get("/{name}")
    fun byName(name: String): Mono<Person> {
        return Mono
            .fromCallable { personService.findByName(name) } // (2)
            .subscribeOn(scheduler) // (3)
    }
}

配置的 I/O 执行器服务被注入
Mono::fromCallable 方法包装阻塞操作
Project Reactor 的 subscribeOn 方法调度 I/O 线程池上的操作

使用@Body 注解

要解析请求正文，您首先要使用 Body 注解向 Micronaut 指示哪个参数接收数据。

以下示例实现了一个简单的回显服务器，它回显请求中发送的正文：

使用@Body 注解

Java Groovy Kotlin

import io.micronaut.http.HttpResponse;
import io.micronaut.http.MediaType;
import io.micronaut.http.annotation.Body;
import io.micronaut.http.annotation.Controller;
import io.micronaut.http.annotation.Post;
import javax.validation.constraints.Size;

@Controller("/receive")
public class MessageController {

@Post(value = "/echo", consumes = MediaType.TEXT_PLAIN) // (1)
String echo(@Size(max = 1024) @Body String text) { // (2)
    return text; // (3)
}

}

import io.micronaut.http.HttpResponse
import io.micronaut.http.MediaType
import io.micronaut.http.annotation.Body
import io.micronaut.http.annotation.Controller
import io.micronaut.http.annotation.Post
import javax.validation.constraints.Size

@Controller("/receive")
class MessageController {

@Post(value = "/echo", consumes = MediaType.TEXT_PLAIN) // (1)
String echo(@Size(max = 1024) @Body String text) { // (2)
    text // (3)
}

}

import io.micronaut.http.HttpResponse
import io.micronaut.http.MediaType
import io.micronaut.http.annotation.Body
import io.micronaut.http.annotation.Controller
import io.micronaut.http.annotation.Post
import javax.validation.constraints.Size

@Controller("/receive")
open class MessageController {

@Post(value = "/echo", consumes = [MediaType.TEXT_PLAIN]) // (1)
open fun echo(@Size(max = 1024) @Body text: String): String { // (2)
    return text // (3)
}

}

Post 注释与 text/plain 的 MediaType 一起使用（默认为 application/json）。
Body 注释与 javax.validation.constraints.Size 一起使用，它将主体的大小限制为最多 1KB。此约束不限制服务器读取/缓冲的数据量。
正文作为方法的结果返回

请注意，读取请求正文是以非阻塞方式完成的，因为请求内容在数据可用时读取并累积到传递给方法的字符串中。

配置文件（例如 application.yml）中的 micronaut.server.maxRequestSize 设置限制了服务器读取/缓冲的数据大小（默认最大请求大小为 10MB）。 @Size 不能替代此设置。

无论限制如何，对于大量数据，将数据累积到内存中的 String 可能会导致服务器内存紧张。更好的方法是在您的项目中包含一个 Reactive 库（例如 Reactor、RxJava 或 Akka），它支持 Reactive 流实现并流式传输它变得可用的数据：

使用 Reactive Streams 读取请求体

Java Groovy Kotlin

import io.micronaut.http.HttpResponse;
import io.micronaut.http.MediaType;
import io.micronaut.http.annotation.Body;
import io.micronaut.http.annotation.Controller;
import io.micronaut.http.annotation.Post;
import javax.validation.constraints.Size;

import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import io.micronaut.core.async.annotation.SingleResult;

@Controller("/receive")
public class MessageController {

@Post(value = "/echo-publisher", consumes = MediaType.TEXT_PLAIN) // (1)
@SingleResult
Publisher<HttpResponse<String>> echoFlow(@Body Publisher<String> text) { //(2)
    return Flux.from(text)
            .collect(StringBuffer::new, StringBuffer::append) // (3)
            .map(buffer -> HttpResponse.ok(buffer.toString()));
}

}

import io.micronaut.http.HttpResponse
import io.micronaut.http.MediaType
import io.micronaut.http.annotation.Body
import io.micronaut.http.annotation.Controller
import io.micronaut.http.annotation.Post
import javax.validation.constraints.Size

import org.reactivestreams.Publisher
import io.micronaut.core.async.annotation.SingleResult
import reactor.core.publisher.Flux

@Controller("/receive")
class MessageController {

@Post(value = "/echo-publisher", consumes = MediaType.TEXT_PLAIN) // (1)
@SingleResult
Publisher<HttpResponse<String>> echoFlow(@Body Publisher<String> text) { // (2)
    return Flux.from(text)
            .collect({ x -> new StringBuffer() }, { StringBuffer sb, String s -> sb.append(s) }) // (3)
            .map({ buffer -> HttpResponse.ok(buffer.toString()) });
}

}

import io.micronaut.http.HttpResponse
import io.micronaut.http.MediaType
import io.micronaut.http.annotation.Body
import io.micronaut.http.annotation.Controller
import io.micronaut.http.annotation.Post
import javax.validation.constraints.Size

import org.reactivestreams.Publisher
import io.micronaut.core.async.annotation.SingleResult
import reactor.core.publisher.Flux

@Controller("/receive")
open class MessageController {

@Post(value = "/echo-publisher", consumes = [MediaType.TEXT_PLAIN]) // (1)
@SingleResult
open fun echoFlow(@Body text: Publisher<String>): Publisher<HttpResponse<String>> { //(2)
    return Flux.from(text)
        .collect({ StringBuffer() }, { obj, str -> obj.append(str) }) // (3)
        .map { buffer -> HttpResponse.ok(buffer.toString()) }
}

}

在这种情况下，该方法被更改为接收和返回 Publisher 类型。
此示例使用 Project Reactor 并返回单个项目。因此，响应类型也使用 SingleResult 进行注释。 Micronaut 只有在操作完成后才会发出响应而不会阻塞。
collect 方法用于在这个模拟示例中累积数据，但它可以将数据逐块写入日志服务、数据库等

不需要转换的类型的主体参数会导致 Micronaut 跳过请求的解码！

Reactive Responses

上一节介绍了使用 Project Reactor 和 Micronaut 进行响应式编程的概念。

Micronaut 支持返回常见的反应类型，例如 Mono（或来自 RxJava 的 Single Maybe Observable 类型）、来自任何控制器方法的 Publisher 或 CompletableFuture 的实例。

要使用 Project Reactor 的 Flux 或 Mono，您需要将 Micronaut Reactor 依赖项添加到您的项目以包含必要的转换器。

要使用 RxJava 的 Flowable、Single 或 Maybe，您需要将 Micronaut RxJava 依赖项添加到您的项目以包含必要的转换器。

使用 Body 注释指定为请求主体的参数也可以是反应类型或 CompletableFuture。

返回响应式类型时，Micronaut 在与请求相同的线程（Netty 事件循环线程）上订阅返回的响应式类型。因此，如果您执行任何阻塞操作，请务必将这些操作卸载到适当配置的线程池中，例如使用 Project Reactor 或 RxJava subscribeOn(..) 工具或@ExecuteOn。

总而言之，下表说明了一些常见的响应类型及其处理：

表 1. Micronaut 响应类型
类型	描述	示例签名
Publisher	任何实现 Publisher 接口的类型	`Publisher<String> hello()`
CompletableFuture	Java CompletableFuture 实例	`CompletableFuture<String> hello()`
HttpResponse	一个 HttpResponse 和可选的响应主体	`HttpResponse<Publisher<String>> hello()`
CharSequence	CharSequence 的任何实现	`String hello()`
T	任何简单的 POJO 类型	`Book show()`

当返回 Reactive 类型时，它的类型会影响返回的响应。例如，当返回一个 Flux 时，Micronaut 无法知道响应的大小，因此使用 Chunked 的 Transfer-Encoding 类型。虽然对于发出单个结果的类型（例如 Mono），会填充 Content-Length 标头。

← Micronaut 接收请求的 Content-Type

Micronaut 用 Jackson 进行 JSON 绑定 →

Micronaut 响应式 HTTP 请求处理

使用@Body 注解

Reactive Responses

推荐文章

推荐教程

最新教程