Reactor 3 Reference Guide

The Reactor reference guide is available as HTML documents. The latest copy is available at https://projectreactor.io/docs/core/release/reference/index.html

Copies of this document may be made for your own use and for distribution to others, provided that you do not charge any fee for such copies and further provided that each copy contains this Copyright Notice, whether distributed in print or electronically.

The reference guide is written in Asciidoc, and you can find its sources at https://github.com/reactor/reactor-core/tree/master/docs/asciidoc.

If you have an improvement or a suggestion, we will be happy to get a pull request from you!

We recommend that you check out a local copy of the repository so that you can generate the documentation by running the asciidoctor gradle task and checking the rendering. Some of the sections rely on included files, so GitHub rendering is not always complete.

You can reach out for help in several ways with Reactor:

Suggest Edit to "About the Documentation"

Reactor 是一个给予JVM实现的完全非堵塞的 reactive 编程基础, 具有高效的 demand 管理 (以 "`backpressure`"的形式管理). 它们直接与Java 8 的 functional APIs 就行了整合, 特别整合了 CompletableFuture, Stream, and Duration. 提供了一系列的可组合的异步 APIs — Flux (为了 [N] 元素) and Mono (为了 [0|1] 元素) — 并大量的实现了 Reactive Streams 规范要求.

Reactor 还支持与给予`reactor-netty` 的项目进行非堵塞的进程间的通信 . 适用于微服务架构, Reactor Netty 为 HTTP (包括 Websockets), TCP, and UDP 提供了支持 backpressure 的网络引擎. Reactive 的编码与解码方式是完全支持.

Reactor Core runs on Java 8 and above.

It has a transitive dependency on org.reactivestreams:reactive-streams:1.0.3.

Reactor 3 uses a BOM (Bill of Materials) model (since reactor-core 3.0.4, with the Aluminium release train). This curated list groups artifacts that are meant to work well together, providing the relevant versions despite potentially divergent versioning schemes in these artifacts.

The BOM is itself versioned, using a release train scheme with a codename followed by a qualifier. The following list shows a few examples:

The codenames represent what would traditionally be the MAJOR.MINOR number. They (mostly) come from the Periodic Table of Elements, in increasing alphabetical order.

The qualifiers are (in chronological order):

As mentioned earlier, the easiest way to use Reactor in your core is to use the BOM and add the relevant dependencies to your project. Note that, when you add such a dependency, you must omit the version so that the version gets picked up from the BOM.

However, if you want to force the use of a specific artifact’s version, you can specify it when adding your dependency, as you usually would. You can also forgo the BOM entirely and specify dependencies by their artifact versions.

Modern applications can reach huge numbers of concurrent users, and, even though the capabilities of modern hardware have continued to improve, performance of modern software is still a key concern.

现代应用程序可以吸引大量的并发用户，即使现代硬件的功能不断提高，现代软件的性能仍然是关键问题。

There are, broadly, two ways one can improve a program’s performance:

广义上讲，有两种方法可以提高程序的性能：

并行使用更多线程和更多硬件资源。

在使用现有资源方面寻求更高的效率。

Usually, Java developers write programs by using blocking code. This practice is fine until there is a performance bottleneck. Then it is time to introduce additional threads, running similar blocking code. But this scaling in resource utilization can quickly introduce contention and concurrency problems.

通常，Java开发人员通过使用阻塞代码来编写程序。除非存在性能瓶颈，否则这种做法很好。 然后是时候引入其他线程，运行类似的阻塞代码了。但是这种资源利用的扩展会迅速引入竞争和并发问题。

Worse still, blocking wastes resources. If you look closely, as soon as a program involves some latency (notably I/O, such as a database request or a network call), resources are wasted because threads (possibly many threads) now sit idle, waiting for data.

更糟糕的是，阻塞会浪费资源。 如果仔细观察，程序一旦遇到一些延迟（特别是I / O，例如数据库请求或网络调用），就会浪费资源， 因为线程（可能有很多线程）现在处于空闲状态，等待数据。

So the parallelization approach is not a silver bullet. It is necessary to access the full power of the hardware, but it is also complex to reason about and susceptible to resource wasting.

因此，并行化方法不是灵丹妙药。有必要访问硬件的全部功能，但是推理和资源浪费也很复杂。

The second approach mentioned earlier, seeking more efficiency, can be a solution to the resource wasting problem. By writing asynchronous, non-blocking code, you let the execution switch to another active task that uses the same underlying resources and later comes back to the current process when the asynchronous processing has finished.

前面提到的第二种方法，寻求更高的效率，可以解决资源浪费的问题。 通过编写异步的非阻塞代码，您可以将执行切换到使用相同基础资源的另一个活动任务， 并在异步处理完成后返回到当前进程。

But how can you produce asynchronous code on the JVM? Java offers two models of asynchronous programming:

但是如何在JVM上生成异步代码？Java提供了两种异步编程模型：

Are these techniques good enough? Not for every use case, and both approaches have limitations.

这些技术够好吗？并非针对每个用例，这两种方法都有局限性。

Callbacks are hard to compose together, quickly leading to code that is difficult to read and maintain (known as “Callback Hell”).

回调很难组合在一起，迅速导致难以阅读和维护的代码（称为“回调地狱”）。

Consider an example: showing the top five favorites from a user on the UI or suggestions if she does not have a favorite. This goes through three services (one gives favorite IDs, the second fetches favorite details, and the third offers suggestions with details), as follows:

考虑一个示例：在用户界面上显示用户的前五个收藏夹，如果没有收藏夹则显示建议。这需要三项服务（一项提供喜欢的ID，第二项获取喜欢的详细信息，第三项提供带有详细信息的建议），如下所示：

That is a lot of code, and it is a bit hard to follow and has repetitive parts. Consider its equivalent in Reactor:

那是很多代码，很难遵循并且包含重复的部分。考虑它在Reactor中的等效功能:

What if you want to ensure the favorite IDs are retrieved in less than 800ms or, if it takes longer, get them from a cache? In the callback-based code, that is a complicated task. In Reactor it becomes as easy as adding a timeout operator in the chain, as follows:

如果要确保在少于800毫秒的时间内检索喜欢的ID，或者如果花费更长的时间我们就从缓存中获取它们，该怎么办？ 在基于回调的代码中，这是一项复杂的任务。在Reactor中，就像在链中添加一个timeout运算符一样容易，如下所示：

Future objects are a bit better than callbacks, but they still do not do well at composition, despite the improvements brought in Java 8 by CompletableFuture. Orchestrating multiple Future objects together is doable but not easy. Also, Future has other problems:

Future对象比回调要好一些，但是尽管Java 8带来了改进，但它们在组合方面仍然表现不佳CompletableFuture。 Future一起编排多个 对象是可行的，但并不容易。另外，Future还有其他问题：

Consider another example: We get a list of IDs from which we want to fetch a name and a statistic and combine these pair-wise, all of it asynchronously. The following example does so with a list of type CompletableFuture:

再看一个例子：我们得到一个ID列表，我们要从中获取一个名称和一个统计信息，并将它们成对组合，所有这些信息都是异步的。 以下示例使用类型列表进行操作CompletableFuture:

Since Reactor has more combination operators out of the box, this process can be simplified, as follows:

由于Reactor提供了更多组合运算符，因此可以简化此过程，如下所示

The perils of using callbacks and Future objects are similar and are what reactive programming addresses with the Publisher-Subscriber pair.

使用回调和Future对象的风险是相似的，这是响应式编程Publisher-Subscriber它们一起要解决的问题

Reactive libraries, such as Reactor, aim to address these drawbacks of “classic” asynchronous approaches on the JVM while also focusing on a few additional aspects:

反应性库（例如Reactor）旨在解决JVM上“经典”异步方法的这些缺点，同时还着重于其他一些方面：

The following image shows how a Flux transforms items:

A Flux<T> is a standard Publisher<T> that represents an asynchronous sequence of 0 to N emitted items, optionally terminated by either a completion signal or an error. As in the Reactive Streams spec, these three types of signal translate to calls to a downstream Subscriber’s onNext, onComplete, and onError methods.

Flux<T>是Publisher<T>代表0到N个发射项目的异步序列的标准，可以选择通过完成信号或错误终止。 如反应式流规范中，这三种类型的信号转换为呼叫到下游Subscriber的onNext，onComplete和onError方法。

With this large scope of possible signals, Flux is the general-purpose reactive type. Note that all events, even terminating ones, are optional: no onNext event but an onComplete event represents an empty finite sequence, but remove the onComplete and you have an infinite empty sequence (not particularly useful, except for tests around cancellation). Similarly, infinite sequences are not necessarily empty. For example, Flux.interval(Duration) produces a Flux<Long> that is infinite and emits regular ticks from a clock.

在这么大范围的可能信号中，Flux是通用信号类型。 请注意，所有事件，甚至是终止事件，都是可选的：没有onNext事件，但一个 onComplete事件表示一个空的有限序列，但是删除onComplete，您将获得一个无限的空序列（除了取消测试外，它不是特别有用）。 同样，无限序列不一定为空。例如，Flux.interval(Duration) 产生一个Flux<Long>无限的并且从时钟发出规则的滴答声。

The following image shows how a Mono transforms an item:

A Mono<T> is a specialized Publisher<T> that emits at most one item and then (optionally) terminates with an onComplete signal or an onError signal.

Mono<T>是一个专门的Publisher<T>，最多发出一个项目， 然后（可选）以一个onComplete信号或一个onError信号终止。

It offers only a subset of the operators that are available for a Flux, and some operators (notably those that combine the Mono with another Publisher) switch to a Flux. For example, Mono#concatWith(Publisher) returns a Flux while Mono#then(Mono) returns another Mono.

它仅提供可用于Flux的运算符的一个子集，而某些运算符（特别是那些Mono与另一个结合的运算符Publisher）切换到Flux。 例如，Mono#concatWith(Publisher)返回Flux一会儿Mono#then(Mono) 返回另一个Mono。

Note that you can use a Mono to represent no-value asynchronous processes that only have the concept of completion (similar to a Runnable). To create one, you can use an empty Mono<Void>.

请注意，您可以使用 Mono表示仅具有完成概念的无值异步流程（类似于Runnable）。要创建一个，您可以使用empty Mono<Void>。

The easiest way to get started with Flux and Mono is to use one of the numerous factory methods found in their respective classes.

最简单的方式开始使用Flux和Mono是使用在各自的类别中的众多工厂方法之一。

For instance, to create a sequence of String, you can either enumerate them or put them in a collection and create the Flux from it, as follows:

例如，要创建的序列String，您可以枚举它们或将它们放入集合中并从中创建Flux，如下所示：

Other examples of factory methods include the following:

When it comes to subscribing, Flux and Mono make use of Java 8 lambdas. You have a wide choice of .subscribe() variants that take lambdas for different combinations of callbacks, as shown in the following method signatures:

当谈到订阅，Flux以及Mono 采用Java 8 lambda表达式。 您可以选择多种.subscribe()类型的变体，它们针对不同的回调组合采用lambda表达式，如以下方法签名所示：

In this section, we introduce the creation of a Flux or a Mono by programmatically defining its associated events (onNext, onError, and onComplete). All these methods share the fact that they expose an API to trigger the events that we call a sink. There are actually a few sink variants, which we’ll get to shortly.

在本节中，我们介绍的创建Flux或Mono通过编程方式定义及其相关事件（onNext，onError，和 onComplete）。 所有这些方法都满足相同的设计：它们公开一个API来触发我们称为接收器的事件。实际上有一些接收器变体，稍后我们将介绍。

Reactor, like RxJava, can be considered to be concurrency-agnostic. That is, it does not enforce a concurrency model. Rather, it leaves you, the developer, in command. However, that does not prevent the library from helping you with concurrency.

像RxJava一样，Reactor可以被视为与并发无关的。 也就是说，它不强制执行并发模型。相反，它使您（开发人员）处于命令状态。但是，这不会阻止你使用并发库。

Obtaining a Flux or a Mono does not necessarily mean that it runs in a dedicated Thread. Instead, most operators continue working in the Thread on which the previous operator executed. Unless specified, the topmost operator (the source) itself runs on the Thread in which the subscribe() call was made. The following example runs a Mono in a new thread:

获得 Flux或 Mono不一定意味着它运行在专用的Thread 中。取而代之的是，大多数运算符Thread将在先前的运算符执行时继续工作。 除非另有说明，最上面的操作（源）本身上运行，Thread其在subscribe()有人呼吁。以下示例Mono在新线程中运行：

The preceding code produces the following output:

In Reactor, the execution model and where the execution happens is determined by the Scheduler that is used. A Scheduler has scheduling responsibilities similar to an ExecutorService, but having a dedicated abstraction lets it do more, notably acting as a clock and enabling a wider range of implementations (virtual time for tests, trampolining or immediate scheduling, and so on).

在Reactor中，执行模型以及执行的位置由所Scheduler使用所确定的 。 Scheduler 具有与ExecutorService相似的调度职责，但是具有专用的抽象使其可以做更多的事情， 尤其是充当时钟并支持更广泛的实现（测试的虚拟时间，蹦床或即时调度等）。

The Schedulers class has static methods that give access to the following execution contexts: Schedulers 类有给访问以下执行上下文的静态方法：

Additionally, you can create a Scheduler out of any pre-existing ExecutorService by using Schedulers.fromExecutorService(ExecutorService). (You can also create one from an Executor, although doing so is discouraged.)

此外，您可以创建一个Scheduler不存在ExecutorService的对象使用Schedulers.fromExecutorService(ExecutorService)。 （Executor尽管不建议这样做，也可以从中创建一个 。）

You can also create new instances of the various scheduler types by using the newXXX methods. For example, Schedulers.newParallel(yourScheduleName) creates a new parallel scheduler named yourScheduleName.

您还可以使用这些newXXX 方法来创建各种调度程序类型的新实例。 例如，Schedulers.newParallel(yourScheduleName)创建一个名为 yourScheduleName 的新并行调度程序。

Some operators use a specific scheduler from Schedulers by default (and usually give you the option of providing a different one). For instance, calling the Flux.interval(Duration.ofMillis(300)) factory method produces a Flux<Long> that ticks every 300ms. By default, this is enabled by Schedulers.parallel(). The following line changes the Scheduler to a new instance similar to Schedulers.single():

某些操作员Schedulers默认情况下使用特定的调度程序（通常会为您提供提供其他调度程序的选项）。 例如，调用 Flux.interval(Duration.ofMillis(300))factory方法会产生一个Flux<Long>每300毫秒滴答一次的滴答声。 默认情况下，通过启用Schedulers.parallel()。以下行将Scheduler更改为类似于Schedulers.single()的新实例：

Reactor offers two means of switching the execution context (or Scheduler) in a reactive chain: publishOn and subscribeOn. Both take a Scheduler and let you switch the execution context to that scheduler. But the placement of publishOn in the chain matters, while the placement of subscribeOn does not. To understand that difference, you first have to remember that nothing happens until you subscribe.

Reactor提供了两种在反应式链中切换执行上下文（或 Scheduler）的方式：publishOn和subscribeOn。两者都使用Scheduler，让您将执行上下文切换到该调度程序。 但是publishOn在链中的位置很重要，而在链中的位置subscribeOn并不重要。要了解这种差异，您首先必须记住，nothing happens until yousubscribe。

In Reactor, when you chain operators, you can wrap as many Flux and Mono implementations inside one another as you need. Once you subscribe, a chain of Subscriber objects is created, backward (up the chain) to the first publisher. This is effectively hidden from you. All you can see is the outer layer of Flux (or Mono) and Subscription, but these intermediate operator-specific subscribers are where the real work happens.

在Reactor中，当您链接运算符时，可以根据需要将许多 实现Flux和Mono实现彼此包装在一起。 订阅后，一个Subscriber对象链将被创建 ，向后（向上）到第一个发布者。 这实际上对您是隐藏的。您所看到的只是Flux（和Mono）和 Subscription的外层，但是这些中间操作员特定的订户才是真正工作的地方。

With that knowledge, we can have a closer look at the publishOn and subscribeOn operators:

有了这些知识，我们可以更详细地了解publishOnand subscribeOn 运算符

In Reactive Streams, errors are terminal events. As soon as an error occurs, it stops the sequence and gets propagated down the chain of operators to the last step, the Subscriber you defined and its onError method.

在反应式流中，错误是终端事件。一旦发生错误，它就会停止序列，并沿操作链传播到最后一步Subscriber上，你定义的onError方法。

Such errors should still be dealt with at the application level. For instance, you might display an error notification in a UI or send a meaningful error payload in a REST endpoint. For this reason, the subscriber’s onError method should always be defined.

此类错误仍应在应用程序级别处理。例如，您可能在UI中显示错误通知，或在REST端点中发送有意义的错误有效负载。 因此，onError应始终定义订户的方法。

Reactor also offers alternative means of dealing with errors in the middle of the chain, as error-handling operators. The following example shows how to do so:

作为错误处理运算符，Reactor还提供了处理链中间错误的替代方法。以下示例显示了如何执行此操作：

Now we can consider each means of error handling one-by-one. When relevant, we make a parallel with imperative programming’s try patterns.

现在，我们可以考虑各种错误处理方式。我们看做与命令式编程的try模式相平行。

Processors are a special kind of Publisher that are also a Subscriber. That means that you can subscribe to a Processor (generally, they implement Flux), but you can also call methods to manually inject data into the sequence or terminate it.

处理器是一种特殊的Publisher，同时也是一个Subscriber。 这意味着你可以subscribe到Processor（通常，他们实施Flux），但你也可以手动调用方法来注入数据的序列或终止它。

There are several kinds of Processors, each with a few particular semantics, but before you start looking into these, you need to ask yourself the following question:

处理器有几种，每种都有一些特殊的语义，但是在开始研究它们之前，您需要问自己以下问题：

Reactor supports Kotlin 1.1+ and requires kotlin-stdlib (or one of its kotlin-stdlib-jre7 or kotlin-stdlib-jre8 variants).

Thanks to its great Java interoperability and to Kotlin extensions, Reactor Kotlin APIs leverage regular Java APIs and are additionally enhanced by a few Kotlin-specific APIs that are available out of the box within Reactor artifacts.

For example, Kotlin reified type parameters provide a workaround for JVM generics type erasure, and Reactor provides some extensions to take advantage of this feature.

The following table compares Reactor with Java against Reactor with Kotlin and extensions:

The Reactor KDoc API lists and documents all the available Kotlin extensions.

One of Kotlin’s key features is null safety, which cleanly deals with null values at compile time rather than bumping into the famous NullPointerException at runtime. This makes applications safer through nullability declarations and expressive “value or no value” semantics without paying the cost of wrappers such as Optional. (Kotlin allows using functional constructs with nullable values. See this comprehensive guide to Kotlin null-safety.)

Although Java does not let one express null safety in its type-system, Reactor now provides null safety of the whole Reactor API through tooling-friendly annotations declared in the reactor.util.annotation package. By default, types from Java APIs used in Kotlin are recognized as platform types for which null-checks are relaxed. Kotlin support for JSR 305 annotations and Reactor nullability annotations provide null-safety for the whole Reactor API to Kotlin developers, with the advantage of dealing with null-related issues at compile time.

You can configure the JSR 305 checks by adding the -Xjsr305 compiler flag with the following options: -Xjsr305={strict|warn|ignore}.

For kotlin versions 1.1.50+, the default behavior is the same as -Xjsr305=warn. The strict value is required to have the Reactor API full null-safety taken into account but should be considered experimental, since the Reactor API nullability declaration could evolve even between minor releases, as more checks may be added in the future).

Suggest Edit to "Kotlin support"

The most common case for testing a Reactor sequence is to have a Flux or a Mono defined in your code (for example, it might be returned by a method) and to want to test how it behaves when subscribed to.

This situation translates well to defining a “test scenario,” where you define your expectations in terms of events, step-by-step. You can ask and answer questions such as the following:

You can express all of that through the StepVerifier API.

For instance, you could have the following utility method in your codebase that decorates a Flux:

In order to test it, you want to verify the following scenario:

In the StepVerifier API, this translates to the following test:

The API is a builder. You start by creating a StepVerifier and passing the sequence to be tested. This offers a choice of methods that let you:

For terminal events, the corresponding expectation methods (expectComplete() and expectError() and all their variants) switch to an API where you cannot express expectations anymore. In that last step, all you can do is perform some additional configuration on the StepVerifier and then trigger the verification, often with verify() or one of its variants.

What happens at this point is that the StepVerifier subscribes to the tested Flux or Mono and plays the sequence, comparing each new signal with the next step in the scenario. As long as these match, the test is considered a success. As soon as there is a discrepancy, an AssertionError is thrown.

Note that, if one of the lambda-based expectations throws an AssertionError, it is reported as is, failing the test. This is useful for custom assertions.

You can use StepVerifier with time-based operators to avoid long run times for corresponding tests. You can do so through the StepVerifier.withVirtualTime builder.

It looks like the following example:

This virtual time feature plugs in a custom Scheduler in Reactor’s Schedulers factory. Since these timed operators usually use the default Schedulers.parallel() scheduler, replacing it with a VirtualTimeScheduler does the trick. However, an important prerequisite is that the operator be instantiated after the virtual time scheduler has been activated.

To increase the chances that this happens correctly, the StepVerifier does not take a simple Flux as input. withVirtualTime takes a Supplier, which guides you into lazily creating the instance of the tested flux after having done the scheduler set up.

There are two expectation methods that deal with time, and they are both valid with or without virtual time:

Both methods pause the thread for the given duration in classic mode and advance the virtual clock instead in virtual mode.

In order to quickly evaluate the behavior of our Mono.delay above, we can finish writing our code as follows:

We could have used thenAwait(Duration.ofDays(1)) above, but expectNoEvent has the benefit of guaranteeing that nothing happened earlier than it should have.

Note that verify() returns a Duration value. This is the real-time duration of the entire test.

After having described the final expectation of your scenario, you can switch to a complementary assertion API instead of triggering verify(). To do so, use verifyThenAssertThat() instead.

verifyThenAssertThat() returns a StepVerifier.Assertions object, which you can use to assert a few elements of state once the whole scenario has played out successfully (because it also calls verify()). Typical (albeit advanced) usage is to capture elements that have been dropped by some operator and assert them (see the section on Hooks).

For more information about the Context, see Adding a Context to a Reactive Sequence.

StepVerifier comes with a couple of expectations around the propagation of a Context:

Additionally, you can associate a test-specific initial Context to a StepVerifier by using StepVerifierOptions to create the verifier.

These features are demonstrated in the following snippet:

For more advanced test cases, it might be useful to have complete mastery over the source of data, to trigger finely chosen signals that closely match the particular situation you want to test.

Another situation is when you have implemented your own operator and you want to verify how it behaves with regards to the Reactive Streams specification, especially if its source is not well behaved.

For both cases, reactor-test offers the TestPublisher class. This is a Publisher<T> that lets you programmatically trigger various signals:

You can get a well behaved TestPublisher through the create factory method. Also, you can create a misbehaving TestPublisher by using the createNonCompliant factory method. The latter takes a value or multiple values from the TestPublisher.Violation enum. The values define which parts of the specification the publisher can overlook. These enum values include:

Finally, the TestPublisher keeps track of internal state after subscription, which can be asserted through its various assert* methods.

You can use it as a Flux or Mono by using the conversion methods, flux() and mono().

When building complex chains of operators, you could come across cases where there are several possible execution paths, materialized by distinct sub-sequences.

Most of the time, these sub-sequences produce a specific-enough onNext signal that you can assert that it was executed by looking at the end result.

For instance, consider the following method, which builds a chain of operators from a source and uses a switchIfEmpty to fall back to a particular alternative if the source is empty:

You can test which logical branch of the switchIfEmpty was used, as follows:

However, think about an example where the method produces a Mono<Void> instead. It waits for the source to complete, performs an additional task, and completes. If the source is empty, a fallback Runnable-like task must be performed instead. The following example shows such a case:

To verify that your processOrFallback method does indeed go through the doWhenEmpty path, you need to write a bit of boilerplate. Namely you need a Mono<Void> that:

Before version 3.1, you would need to manually maintain one AtomicBoolean per state you wanted to assert and attach a corresponding doOn* callback to the publisher you wanted to evaluate. This could be a lot of boilerplate when having to apply this pattern regularly. Fortunately, 3.1.0 introduced an alternative with PublisherProbe. The following example shows how to use it:

You can also use the probe in place of a Flux<T> by calling .flux() instead of .mono(). For cases where you need to probe an execution path but also need the probe to emit data, you can wrap any Publisher<T> by using PublisherProbe.of(Publisher).

Suggest Edit to "Testing"

When you shift to asynchronous code, things can get much more complicated.

当您转向异步代码时，事情会变得更加复杂。

Consider the following stack trace:

考虑以下堆栈跟踪：

典型的Reactor堆栈跟踪

There is a lot going on there. We get an IndexOutOfBoundsException, which tells us that a source emitted more than one item.

那里发生了很多事情。我们得到一个IndexOutOfBoundsException，告诉我们一个`source emitted more than one item`.

We can probably quickly come to assume that this source is a Flux or a Mono, as confirmed by the next line, which mentions MonoSingle. So it appears to be some sort of complaint from a single operator.

正如下一行提到的所示，我们可能很快就可以假定此源是Flux或Mono MonoSingle。因此，这似乎是single操作员的某种抱怨。

Referring to the Javadoc for the Mono#single operator, we see that single has a contract: The source must emit exactly one element. It appears we had a source that emitted more than one and thus violated that contract.

参考Mono#single操作符的Javadoc时，我们看到single有一个约定：源必须确切地发出一个元素。看来我们有一个排放源不止一个，因此违反了该合同。

Can we dig deeper and identify that source? The following rows are not very helpful. They take us through the internals of what seems to be a reactive chain, through multiple calls to subscribe and request.

我们可以更深入地挖掘并确定来源吗？以下几行不是很有帮助。 带我们进入似乎是一个reactive 链的内部。进过了多次调用subscribe和request，

By skimming over these rows, we can at least start to form a picture of the kind of chain that went wrong: It seems to involve a MonoSingle, a FluxFlatMap, and a FluxRange (each gets several rows in the trace, but overall these three classes are involved). So a range().flatMap().single() chain maybe?

通过略读这些行，我们至少可以开始形成一条错误链的图：似乎涉及到MonoSingle，a FluxFlatMap和a FluxRange （每个都在跟踪中得到几行，但总体上涉及这三个类）。 那么range().flatMap().single()也许是一个调用链？

But what if we use that pattern a lot in our application? This still does not tell us much, and simply searching for single is not going to find the problem. Then the last line refers to some of our code. Finally, we are getting close.

但是，如果我们在应用程序中频繁使用该模式怎么办？这仍然不能告诉我们太多，仅搜索single是不会发现问题的。 然后，最后一行引用了我们的一些代码。最后，我们越来越近了。

Hold on, though. When we go to the source file, all we see is that a pre-existing Flux is subscribed to, as follows:

等一下。转到源文件时，我们看到 之前的退出的Flux 是已被预订了，如下所示：

All of this happened at subscription time, but the Flux itself was not declared there. Worse, when we go to where the variable is declared, we see the following:

所有这些都是在订阅时发生的，但Flux本身并未在此处声明。更糟糕的是，当我们转到声明变量的位置时，会看到以下内容：

The variable is not instantiated where it is declared. We must assume a worst-case scenario where we find out that there could be a few different code paths that set it in the application. We remain unsure of which one caused the issue.

变量未在声明的地方实例化。 我们必须假设在最坏的情况下，我们发现可能在应用程序中设置了一些不同的代码路径。我们仍然不确定是哪一个引起了问题。

What we want to find out more easily is where the operator was added into the chain - that is, where the Flux was declared. We usually refer to that as the “assembly” of the Flux.

我们想要更容易发现的是将运算符添加到链中的位置-即Flux声明的位置。我们通常将其称为Flux的“assembly” 。

Even though the stacktrace was still able to convey some information for someone with a bit of experience, we can see that it is not ideal by itself in more advanced cases.

即使stacktrace仍然能够为有经验的人传达一些信息，但我们可以看到，在更高级的情况下，它本身并不理想。

Fortunately, Reactor comes with assembly-time instrumentation that is designed for debugging.

幸运的是，Reactor带有专为调试而设计的组装时工具。

This is done by customizing the Hooks.onOperator hook at application start (or at least before the incriminated Flux or Mono can be instantiated), as follows:

这是通过Hooks.onOperator在应用程序启动时（或至少在包含Flux或Mono可实例化之前）自定义钩子来完成的，如下所示：

This starts instrumenting the calls to the Flux (and Mono) operator methods (where they are assembled into the chain) by wrapping the construction of the operator and capturing a stack trace there. Since this is done when the operator chain is declared, the hook should be activated before that, so the safest way is to activate it right at the start of your application.

通过包装操作符的结构并捕获其中的堆栈跟踪信息，开始对Flux（和Mono）操作符方法（将它们组装到链中）的调用进行检测。 由于此操作是在声明操作员链时完成的，因此应在此之前将钩子激活，因此最安全的方法是在应用程序开始时立即将其激活。

Later on, if an exception occurs, the failing operator is able to refer to that capture and append it to the stack trace. We call this captured assembly information a traceback.

稍后，如果发生异常，则失败的运算符可以引用该捕获并将其附加到堆栈跟踪中。我们将此捕获的程序集信息称为回溯。

In the next section, we see how the stack trace differs and how to interpret that new information.

在下一节中，我们将了解堆栈跟踪的不同之处以及如何解释该新信息。

When we reuse our initial example but activate the operatorStacktrace debug feature, the stack trace is as follows:

当我们重用最初的示例但激活operatorStacktrace调试功能时，堆栈跟踪如下：

The captured stack trace is appended to the original error as a suppressed OnAssemblyException. There are two parts to it, but the first section is the most interesting. It shows the path of construction for the operator that triggered the exception. Here, it shows that the single that caused our issue was created in the scatterAndGather method, itself called from a populateDebug method that got executed through JUnit.

捕获的堆栈跟踪将被抑制后附加到原始错误OnAssemblyException。它有两个部分，但是第一部分是最有趣的。它显示了触发异常的操作员的构造路径。 在这里，它表明single导致我们问题的是在scatterAndGather方法中创建的，该 方法本身是populateDebug通过JUnit执行的方法调用的。

Now that we are armed with enough information to find the culprit, we can have a meaningful look at that scatterAndGather method:

既然我们已经掌握了足够的信息来找到罪魁祸首，我们就可以对该scatterAndGather方法进行有自信的排查：

Now we can see what the root cause of the error was a flatMap that performs several HTTP calls to a few URLs but that is chained with single, which is too restrictive. After a short git blame and a quick discussion with the author of that line, we find out he meant to use the less restrictive take(1) instead.

现在，我们可以看到错误的根本原因是flatMap它对几个URL执行了几次HTTP调用，但是与链接在一起single，这太过严格了。 在git blame与该行的作者进行简短简短的讨论之后，我们发现他的意思是使用限制性较小的take(1)标签。

We have solved our problem.

我们已经解决了我们的问题。

Now consider the following line in the stack trace:

现在考虑堆栈跟踪中的以下行：

That second part of the debug stack trace was not necessarily interesting in this particular example, because the error was actually happening in the last operator in the chain (the one closest to subscribe). Considering another example might make it more clear:

在此特定示例中，调试堆栈跟踪的第二部分不一定是有趣的，因为该错误实际上发生在链中的最后一个运算符中（最接近的那个subscribe）。 考虑另一个示例可能会更清楚：

Now imagine that, inside findAllUserByName, there is a map that fails. Here, we would see the following final traceback:

现在想象一下，在内部findAllUserByName，有一个map失败。在这里，我们将看到以下最终回溯：

This corresponds to the section of the chain of operators that gets notified of the error:

这对应于操作员链中获得该错误通知的部分：

We deal with a form of instrumentation here, and creating a stack trace is costly. That is why this debugging feature should only be activated in a controlled manner, as a last resort.

我们在这里处理一种形式的检测，并且创建堆栈跟踪非常昂贵。这就是为什么只能以可控制的方式激活此调试功能的原因。

Project Reactor comes with a separate Java Agent that instruments your code and adds debugging info without paying the cost of capturing the stacktrace on every operator call. The behaviour is very similar to Activating Debug Mode - aka tracebacks, but without the runtime performance overhead.

Project Reactor带有一个单独的Java代理，可对您的代码进行检测并添加调试信息，而无需话费每次操作调用时捕获stacktrace的消耗。 该行为与 Activating Debug Mode - aka tracebacks（也称为回溯）非常相似，但没有运行时性能开销。

To use it in your app, you must add it as a dependency.

要在您的应用程序中使用它，必须将其添加为依赖项。

The following example shows how to add reactor-tools as a dependency in Maven:

下面的示例显示如何reactor-tools在Maven中添加为依赖项：

The following example shows how to add reactor-tools as a dependency in Gradle:

It also needs to be explicitly initialized with:

还需要使用以下命令显式初始化它：

You may also re-process existing classes if you cannot run the init eagerly (e.g. in the tests):

如果您不能急于运行init（例如在测试中），也可以重新处理现有的类：

In addition to stack trace debugging and analysis, another powerful tool to have in your toolkit is the ability to trace and log events in an asynchronous sequence.

除了堆栈跟踪调试和分析之外，工具包中另一个强大的工具是能够以异步顺序跟踪和记录事件。

The log() operator can do just that. Chained inside a sequence, it peeks at every event of the Flux or Mono upstream of it (including onNext, onError, and onComplete as well as subscriptions, cancellations, and requests).

For instance, suppose we have Logback activated and configured and a chain like range(1,10).take(3). By placing a log() before the take, we can get some insight into how it works and what kind of events it propagates upstream to the range, as the following example shows:

例如，假设我们已激活并配置了Logback以及类似的链 range(1,10).take(3)。 通过在take之前放置log()，我们可以深入了解其工作原理以及它将向上游传播到范围的事件的种类，如以下示例所示：

This prints out the following (through the logger’s console appender):

这将打印出以下内容（通过记录器的控制台附加）：

The last line, (4), is the most interesting. We can see the take in action there. It operates by cutting the sequence short after it has seen enough elements emitted. In short, take() causes the source to cancel() once it has emitted the user-requested amount.

最后一行（4）是最有趣的。我们可以take在那里看到实际的效果。 在看到足够多的元素发射之后，它会通过缩短序列来进行操作。简而言之，take()使源转变为cancel()的原因是让源发送完一次用户请求的数量。

Suggest Edit to "Debugging Reactor"

Every async operation in Reactor is done via the Scheduler abstraction described in Threading and Schedulers. This is why it is important to monitor your schedulers, watch out for key metrics that start to look suspicious and react accordingly.

To enable scheduler metrics, you will need to use the following method:

Once scheduler metrics are enabled and provided it is on the classpath, Reactor will use Micrometer’s support for instrumenting the executors that back most schedulers.

Please refer to Micrometer’s documentation for the exposed metrics, such as:

Since one scheduler may have multiple executors, every executor metric has a reactor_scheduler_id tag.

Sometimes it is useful to be able to record metrics at some stage in your reactive pipeline.

One way to do it would be to manually push the values to your metrics backend of choice. Another option would be to use Reactor’s built-in metrics integration for Flux/Mono and interpret them.

Consider the following pipeline:

To enable the metrics for this source Flux (returned from listenToEvents()), we need to give it a name and turn on the metrics collecting:

Just adding these two operators will expose a whole bunch of useful metrics!

Want to know how many times your event processing has restarted due to some error? Read reactor.subscribed, because retry() operator will re-subscribe to the source publisher on error.

Interested in "events per second" metric? Measure the rate of reactor.onNext.delay 's count.

Want to be alerted when the listener throws an error? reactor.flow.duration with status=error tag is your friend.

From a clean-code perspective, code reuse is generally a good thing. Reactor offers a few patterns that can help you reuse and mutualize code, notably for operators or combinations of operators that you might want to apply regularly in your codebase. If you think of a chain of operators as a recipe, you can create a “cookbook” of operator recipes.

从干净代码的角度来看，代码重用通常是一件好事。Reactor提供了一些模式，可以帮助您重用和互用代码，特别是对于您可能希望在代码库中定期应用的运算符或运算符组合。 如果您将一连串的操作符视为配方，则可以创建一个“菜谱”操作符配方。

So far, we have considered that all Flux (and Mono) are the same: They all represent an asynchronous sequence of data, and nothing happens before you subscribe.

到目前为止，我们已经考虑到所有Flux（和Mono）都是相同的：它们都代表异步的数据序列，在您订阅之前什么也没有发生。

Really, though, there are two broad families of publishers: hot and cold.

不过，实际上，有两个广泛的发布者家族：热门和冷门。

The earlier description applies to the cold family of publishers. They generate data anew for each subscription. If no subscription is created, data never gets generated.

较早的描述适用于冷门的发布者系列。它们为每个订阅重新生成数据。如果没有创建订阅，则永远不会生成数据。

Think of an HTTP request: Each new subscriber triggers an HTTP call, but no call is made if no one is interested in the result.

考虑一下HTTP请求：每个新订户都会触发HTTP调用，但是如果没有人对结果感兴趣，则不会进行任何调用。

Hot publishers, on the other hand, do not depend on any number of subscribers. They might start publishing data right away and would continue doing so whenever a new Subscriber comes in (in which case, the subscriber would see only new elements emitted after it subscribed). For hot publishers, something does indeed happen before you subscribe.

另一方面，热门发布者不依赖任何数量的订阅者。他们可能会开始发布数据的时候了，并会继续这样做，每当有新 Subscriber进来（在这种情况下，用户会看到其预约后发射的新元素）。 对于热门发布者，在您订阅之前确实发生了某些事情。

One example of the few hot operators in Reactor is just: It directly captures the value at assembly time and replays it to anybody subscribing to it later. To re-use the HTTP call analogy, if the captured data is the result of an HTTP call, then only one network call is made, when instantiating just.

在Reactor中为数不多的热门运算符的一个示例是just：它直接在汇编时捕获值，然后将其重播给以后订阅该值的任何人。 为了重用HTTP调用类比，如果捕获的数据是HTTP调用的结果，则在实例化时仅进行一个网络调用just。

To transform just into a cold publisher, you can use defer. It defers the HTTP request in our example to subscription time (and would result in a separate network call for each new subscription).

要转化just为冷发行商，您可以使用defer。它在我们的示例中将HTTP请求延迟了订阅时间（并会为每个新订阅导致单独的网络调用）。

Consider two other examples. The following code shows the first example:

考虑另外两个例子。以下代码显示了第一个示例：

This first example produces the following output:

The following image shows the replay behavior:

Both subscribers catch all four colors, because each subscriber causes the process defined by the operators on the Flux to run.

两个订户都捕获全部四种颜色，因为每个订户都会导致Flux运行由操作符定义的过程。

Compare the first example to the second example, shown in the following code:

将第一个示例与第二个示例进行比较，如以下代码所示：

The second example produces the following output:

The following image shows how a subscription is broadcast:

Subscriber 1 catches all four colors. Subscriber 2, having been created after the first two colors were produced, catches only the last two colors. This difference accounts for the doubling of ORANGE and PURPLE in the output. The process described by the operators on this Flux runs regardless of when subscriptions have been attached.

订户1捕获所有四种颜色。在生成前两种颜色之后创建的订户2，仅捕获后两种颜色。 这种差异导致输出ORANGE和的加倍PURPLE。无论何时附加订阅，运营商在此Flux上描述的过程都将运行。

Sometimes, you may want to not defer only some processing to the subscription time of one subscriber, but you might actually want for several of them to rendezvous and then trigger the subscription and data generation.

有时，您可能不希望仅将某些处理推迟到一个订户的订阅时间，但实际上您可能希望其中的几个会合，然后触发订阅和数据生成。

This is what ConnectableFlux is made for. Two main patterns are covered in the Flux API that return a ConnectableFlux: publish and replay.

这是ConnectableFlux能做的。Flux API 中涵盖了两个主要模式，这些模式返回ConnectableFlux：publish和replay。

A ConnectableFlux offers additional methods to manage subscriptions downstream versus subscriptions to the original source. These additional methods include the following:

ConnectableFlux提供了其他方法来管理下游订阅，而不是原始来源的订阅。这些其他方法包括：

Consider the following example:

The preceding code produces the following output:

The following code uses autoConnect:

The preceding code produces the following output:

When you have lots of elements and you want to separate them into batches, you have three broad solutions in Reactor: grouping, windowing, and buffering. These three are conceptually close, because they redistribute a Flux<T> into an aggregate. Grouping and windowing create a Flux<Flux<T>>, while buffering aggregates into a Collection<T>.

当您有很多元素并且想要将它们分成批处理时，Reactor中提供了三种广泛的解决方案：分组，窗口化和缓冲。这三个在概念上很接近，因为它们将重新分配Flux<T>为一个集合。 分组和开窗创建一个Flux<Flux<T>>，同时将聚合缓冲到一个Collection<T>。

With multi-core architectures being a commodity nowadays, being able to easily parallelize work is important. Reactor helps with that by providing a special type, ParallelFlux, that exposes operators that are optimized for parallelized work.

如今，随着多核体系结构成为一种商品，能够轻松并行化工作非常重要。 Reactor通过提供一种特殊类型来帮助实现这一点，该类型 ParallelFlux公开了针对并行工作进行了优化的运算符。

To obtain a ParallelFlux, you can use the parallel() operator on any Flux. By itself, this method does not parallelize the work. Rather, it divides the workload into “rails” (by default, as many rails as there are CPU cores).

要获取 ParallelFlux，您可以parallel()在任意位置使用运算符Flux。就其本身而言，此方法不会使工作并行化。 相反，它将工作负载划分为“ rails”（默认情况下，与CPU内核一样多的rails）。

In order to tell the resulting ParallelFlux where to run each rail (and, by extension, to run rails in parallel) you have to use runOn(Scheduler). Note that there is a recommended dedicated Scheduler for parallel work: Schedulers.parallel().

为了告诉最终结果ParallelFlux，每个导轨都在哪里运行（并扩展为并行运行导轨），您必须使用runOn(Scheduler)。 需要注意的是有一个推荐的专用Scheduler并行工作：Schedulers.parallel()。

Compare the next two examples:

The first example produces the following output:

The second correctly parallelizes on two threads, as shown in the following output:

第二个在两个线程上正确并行化，如以下输出所示：

If, once you process your sequence in parallel, you want to revert back to a “normal” Flux and apply the rest of the operator chain in a sequential manner, you can use the sequential() method on ParallelFlux.

如果在并行处理序列后想要恢复为“正常” Flux并以顺序方式应用其余运算符链，则可以在ParallelFlux上使用sequential()方法。

Note that sequential() is implicitly applied if you subscribe to the ParallelFlux with a Subscriber but not when using the lambda-based variants of subscribe.

请注意，sequential()是隐式应用，如果你subscribe到ParallelFlux 了Subscriber使用基于lambda时，但不是subscribe。

Note also that subscribe(Subscriber<T>) merges all the rails, while subscribe(Consumer<T>) runs all the rails. If the subscribe() method has a lambda, each lambda is executed as many times as there are rails.

还要注意，subscribe(Subscriber<T>)合并所有导轨，同时 subscribe(Consumer<T>)运行所有导轨。 如果该subscribe()方法具有lambda，则每个lambda的执行次数将与rails一样多。

You can also access individual rails or “groups” as a Flux<GroupedFlux<T>> through the groups() method and apply additional operators to them through the composeGroup() method.

您也可以Flux<GroupedFlux<T>>通过groups()方法访问单个导轨或“组”，并通过该 方法向它们应用其他运算符composeGroup() 。

As we described in the Threading and Schedulers section, Reactor Core comes with several Scheduler implementations. While you can always create new instances through the new* factory methods, each Scheduler flavor also has a default singleton instance that is accessible through the direct factory method (such as Schedulers.boundedElastic() versus Schedulers.newBoundedElastic(…​)).

如我们在“ 线程和调度程序”部分所述，Reactor Core附带了几种 Scheduler实现。 尽管您始终可以通过new* 工厂方法创建新实例，但是每个Scheduler风味还具有默认的单例实例，可通过直接工厂方法（例如Schedulers.boundedElastic()vs Schedulers.newBoundedElastic(…​)）进行访问。

These default instances are the ones used by operators that need a Scheduler to work when you do not explicitly specify one. For example, Flux#delayElements(Duration) uses the Schedulers.parallel() instance.

这些默认实例是Scheduler未明确指定实例时需要工作的运算符所使用的实例。 例如，Flux#delayElements(Duration)使用Schedulers.parallel()实例。

In some cases, however, you might need to change these default instances with something else in a cross-cutting way, without having to make sure every single operator you call has your specific Scheduler as a parameter. An example is measuring the time every single scheduled task takes by wrapping the real schedulers, for instrumentation purposes. In other words, you might want to change the default Schedulers.

但是，在某些情况下，您可能需要以交叉方式使用其他默认值来更改这些默认实例，而不必确保调用的每个运算符都将自己的特定Scheduler参数作为参数。 一个示例是通过包装实际的计划程序来测量每个计划的任务所花费的时间，以进行检测。换句话说，您可能想要更改默认的Schedulers。

Changing the default schedulers is possible through the Schedulers.Factory class. By default, a Factory creates all the standard Scheduler through similarly named methods. You can override each of these with your custom implementation.

通过Schedulers.Factory该类可以更改默认调度程序。 默认情况下，Factory 通过相似命名的方法创建所有标准Scheduler。您可以使用自定义实现覆盖其中的每一个。

Additionally, the factory exposes one additional customization method: decorateExecutorService. It is invoked during the creation of every Reactor Core Scheduler that is backed by a ScheduledExecutorService (even non-default instances, such as those created by calls to Schedulers.newParallel()).

此外，工厂还公开了另一种自定义方法： decorateExecutorService。 在创建每个Scheduler由ScheduledExecutorService（作为非默认实例，例如通过调用创建的实例）支持的Reactor Core的过程中调用它 Schedulers.newParallel()。

This lets you tune the ScheduledExecutorService to be used: The default one is exposed as a Supplier and, depending on the type of Scheduler being configured, you can choose to entirely bypass that supplier and return your own instance or you can get() the default instance and wrap it.

这使您可以调整ScheduledExecutorService要使用的：默认实例显示为一个Supplier并且根据Scheduler配置的类型， 您可以选择完全绕过该提供并返回自己的实例，也可以为get()将选择的默认实例包装并将其返回。

Finally, there is a last customizable hook in Schedulers: onHandleError. This hook is invoked whenever a Runnable task submitted to a Scheduler throws an Exception (note that if there is an UncaughtExceptionHandler set for the Thread that ran the task, both the handler and the hook are invoked).

最后，还有最后一个可定制的钩子Schedulers：onHandleError。 每当Runnable提交给Scheduler引发任务的任务抛出异常时，都会调用此挂钩Exception（请注意，如果有运行该任务的UncaughtExceptionHandler集合，则将Thread同时调用处理程序和挂钩）。

Reactor has another category of configurable callbacks that are invoked by Reactor operators in various situations. They are all set in the Hooks class, and they fall into three categories:

Reactor具有另一类可配置的回调，可在各种情况下由Reactor运算符调用。它们全都设置在Hooks 类中，分为三类：

One of the big technical challenges encountered when switching from an imperative programming perspective to a reactive programming mindset lies in how you deal with threading.

从命令式编程视角转换为反应式编程思维方式时遇到的重大技术挑战之一是如何处理线程。

Contrary to what you might be used to, in reactive programming, you can use a Thread to process several asynchronous sequences that run at roughly the same time (actually, in non-blocking locksteps). The execution can also easily and often jump from one thread to another.

与您习惯的反应式编程相反，您可以使用`Thread`处理几个大致同时运行的异步序列（实际上，非阻塞的锁步）。 执行也可以轻松且经常从一个线程跳转到另一个。

This arrangement is especially hard for developers that use features dependent on the threading model being more “stable,” such as ThreadLocal. As it lets you associate data with a thread, it becomes tricky to use in a reactive context. As a result, libraries that rely on ThreadLocal at least introduce new challenges when used with Reactor. At worst, they work badly or even fail. Using the MDC of Logback to store and log correlation IDs is a prime example of such a situation.

对于使用依赖于线程模型上开发更“稳定”的功能这种安排尤其困难。比如ThreadLocal。因为它使您可以将数据与线程相关联，所以在反应式上下文中使用它变得棘手。 ThreadLocal与Reactor一起使用时，至少依赖的库会带来新的挑战。在最坏的情况下，它们会表现不佳甚至失败。 使用Logback的MDC存储和记录相关ID是这种情况的主要示例。

The usual workaround for ThreadLocal usage is to move the contextual data, C, along your business data, T, in the sequence, by using (for instance) Tuple2<T, C>. This does not look good and leaks an orthogonal concern (the contextual data) into your method and Flux signatures.

使用的通常解决方法ThreadLocal是使用（例如）按顺序移动上下文数据C和业务数据。这看起来不好，并且将正交关注点（上下文数据）泄漏到您的方法和 签名中。TTuple2<T, C>Flux

Since version 3.1.0, Reactor comes with an advanced feature that is somewhat comparable to ThreadLocal but can be applied to a Flux or a Mono instead of a Thread. This feature is called Context.

从版本开始3.1.0，Reactor带有一项先进的功能，在某种程度上可以与ThreadLocal媲美，但可以应用于Flux或Mono代替Thread。此功能称为Context。

As an illustration of what it looks like, the following example both writes from and writes to Context:

为了说明其外观，以下示例同时写入和写入Context：

In the following sections, we cover Context and how to use it, so that you can eventually understand the preceding example.

在以下各节中，我们介绍Context了它以及如何使用它，以便您最终可以理解前面的示例。

In very specific cases, your application may deal with types that necessitate some form of cleanup once they are no longer in use. This is an advanced scenario — for, example when you have reference-counted objects or when you deal with off-heap objects. Netty’s ByteBuf is a prime example of both.

在非常特定的情况下，您的应用程序可能会处理不再需要使用某种形式的清理的类型。这是一种高级方案，例如，当您有引用计数的对象或处理堆外对象时。Netty ByteBuf是这两者的典型例子。

In order to ensure proper cleanup of such objects, you need to account for it on a Flux-by-Flux basis, as well as in several of the global hooks (see Using Global Hooks):

为了确保正确清理此类对象，您需要基于“ Flux”（逐个）“ Flux”以及多个全局挂钩（see Using Global Hooks）进行考虑：

This is needed because each hook is made with a specific subset of cleanup in mind, and users might want (for example) to implement specific error-handling logic in addition to cleanup logic within onOperatorError.

之所以需要这样做，是因为每个挂钩都是根据特定的清理子集来进行的，并且用户可能希望（例如）除中的清理逻辑之外还实现特定的错误处理逻辑onOperatorError。

Note that some operators are less adapted to dealing with objects that need cleanup. For example, bufferWhen can introduce overlapping buffers, and that means that the discard “local hook” we used earlier might see a first buffer as being discarded and cleanup an element in it that is in a second buffer, where it is still valid.

请注意，某些运算符不太适合处理需要清除的对象。 例如，“ bufferWhen”可以引入重叠的缓冲区，这意味着我们前面使用的丢弃“ local hook”可能会看到第一个缓冲区被丢弃，并清理其中第二个缓冲区中的元素。 仍然有效。

Although Java does not allow expressing null-safety with its type system, Reactor now provides annotations to declare nullability of APIs, similar to those provided by Spring Framework 5.

尽管Java不允许使用其类型系统表示空安全性，但是Reactor现在提供了注释来声明API的空性，类似于Spring Framework 5提供的注释。

Reactor uses these annotations, but they can also be used in any Reactor-based Java project to declare null-safe APIs. Nullability of the types used inside method bodies is outside of the scope of this feature.

Reactor使用这些注释，但也可以在任何基于Reactor的Java项目中使用它们来声明空安全的API。 方法主体内使用的类型的可空性超出了此功能的范围。

These annotations are meta-annotated with JSR 305 annotations (a dormant JSR that is supported by tools such as IntelliJ IDEA) to provide useful warnings to Java developers related to null-safety in order to avoid NullPointerException at runtime. JSR 305 meta-annotations let tooling vendors provide null safety support in a generic way, without having to hard-code support for Reactor annotations.

这些注释用JSR 305注释（由IntelliJ IDEA之类的工具支持的休眠JSR）进行元注释，以向Java开发人员提供有关空安全性的有用警告，以避免在运行时出现NullPointerException。 JSR 305元注释使工具供应商能够以通用方式提供空安全支持，而不必对Reactor注释进行硬编码支持。

They are also used by Kotlin, which natively supports null safety. See this dedicated section for more details.

The following annotations are provided in the reactor.util.annotation package:

Suggest Edit to "Advanced Features and Concepts"