1. Introduction

1.绪论

Project Reactor provides a fully non-blocking programming foundation for JVM. It offers implementation of Reactive Streams specification and provides composable asynchronous API such as Flux. A Flux is a Reactive Streams publisher with several reactive operators. It emits 0 to N elements and then completes successfully or with error. It can be created in several different ways depending on our needs.

Project Reactor为JVM提供了一个完全非阻塞的编程基础。它提供了Reactive Streams规范的实现，并提供了可组合的异步API，如Flux。Flux是一个具有多个反应式操作符的反应式流发布器。它发射0到N个元素，然后成功或出错地完成。它可以根据我们的需要以几种不同的方式创建。

2. Understanding Flux

2.了解通量

A Flux is a Reactive Stream publisher that can emit 0 to N elements. It has several operators which are used to generate, orchestrate, and transform Flux sequences. A Flux can complete successfully or complete with errors.

Flux是一个反应式流发布器，可以发布0到N个元素。它有几个运算符，用于生成、协调和转换Flux序列。Flux可以成功完成，也可以带着错误完成。

Flux API provides several static factory methods on Flux to create sources or generate from several callback types. It also provides instance methods and operators to build an asynchronous processing pipeline. This pipeline produces an asynchronous sequence.

Flux API在Flux上提供了几个静态工厂方法，以创建源或从几个回调类型中生成。它还提供了实例方法和操作者来建立一个异步处理管道。这个流水线产生一个异步序列。

In the next sections, let’s see a few usages of the Flux generate() and create() methods.

在接下来的章节中，让我们看看Flux的generate()和create()方法的一些用法。

3. Maven Dependencies

3.Maven的依赖性

We’ll need the reactor-core and reactor-test Maven dependencies:

我们需要reactor-core和reactor-testMaven依赖项。

<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-core</artifactId>
    <version>3.4.17</version>
</dependency>
<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-test</artifactId>
    <version>3.4.17</version>
    <scope>test</scope>
</dependency>

4. Flux Generate

4 通量产生

The generate() method of the Flux API provides a simple and straightforward programmatic approach to creating a Flux. The generate() method takes a generator function to produce a sequence of items.

Flux API的generate()方法为创建Flux提供了一个简单明了的程序化方法。generate()方法需要一个生成器函数来产生一个项目序列。

There are three variants of the generate method:

生成方法有三种变体。

generate(Consumer<SynchronousSink<T>> generator)
generate(Callable<S> stateSupplier, BiFunction<S, SynchronousSink<T>, S> generator)
generate(Callable<S> stateSupplier, BiFunction<S, SynchronousSink<T>, S> generator, Consumer<? super S> stateConsumer)

The generate method calculates and emits the values on demand. It is preferred to use in cases where it is too expensive to calculate elements that may not be used downstream. It can also be used if the emitted events are influenced by the state of the application.

生成方法按需计算并发出数值。在计算可能不会在下游使用的元素的成本太高的情况下，它是首选。如果发射的事件受到应用程序状态的影响，也可以使用该方法。

4.1. Example

4.1.例子

In this example, let use the generate(Callable<S> stateSupplier, BiFunction<S, SynchronousSink<T>, S> generator) to generate a Flux:

在这个例子中，让我们使用generate(Callable<S> stateSupplier, BiFunction<S, SynchronousSink<T> , S> generator)来生成一个Flux。

public class CharacterGenerator {
    
    public Flux<Character> generateCharacters() {
        
        return Flux.generate(() -> 97, (state, sink) -> {
            char value = (char) state.intValue();
            sink.next(value);
            if (value == 'z') {
                sink.complete();
            }
            return state + 1;
        });
    }
}

In the generate() method, we are supplying two functions as arguments:

在generate()方法中，我们提供了两个函数作为参数。

The first one is a Callable function. This function defines the initial state for the generator with the value 97
The second one is a BiFunction. This is a generator function that consumes a SynchronousSink. This SynchronousSink returns an item whenever the sink’s next method is invoked

Based on its name, a SynchronousSink instance works synchronously. However, we cannot call the SynchronousSink object’s next method more than once per generator call.

根据其名称，SynchronousSink实例是同步工作的。然而，我们不能在每个生成器调用中多次调用SynchronousSink对象的next方法。

Let’s verify the generated sequence with StepVerifier:

让我们用StepVerifier验证生成的序列。

@Test
public void whenGeneratingCharacters_thenCharactersAreProduced() {
    CharacterGenerator characterGenerator = new CharacterGenerator();
    Flux<Character> characterFlux = characterGenerator.generateCharacters().take(3);

    StepVerifier.create(characterFlux)
      .expectNext('a', 'b', 'c')
      .expectComplete()
      .verify();
}

In this example, the subscriber requests just three items. Hence the generated sequence ends by emitting three characters – a,b, and c. The expectNext() expects the elements we are expecting from the Flux. The expectComplete() indicates the completion of element emission from the Flux.

在这个例子中，订阅者只要求三个项目。因此，生成的序列以发射三个字符结束–a、b和c。expectNext()期待我们从Flux中期待的元素。expectComplete()表示Flux的元素发射完成。

5. Flux Create

5.通量创建

The create() method in Flux is used when we want to calculate multiple (0 to infinity) values that are not influenced by the application’s state. This is because the underlying method of the Flux create() method keeps calculating the elements.

Flux中的create()方法是在我们想要计算多个（0到无穷大）值时使用的，这些值不受应用程序的状态影响。这是因为Flux的create()方法的底层方法一直在计算这些元素。

Besides, the downstream system determines how many elements it needs. Therefore, if the downstream system is unable to keep up, already emitted elements are either buffered or removed.

此外，下游系统决定它需要多少元素。因此，如果下游系统无法跟上，已经发出的元素就会被缓冲或删除。

By default, the emitted elements are buffered until the downstream system request more elements.

默认情况下，发射的元素被缓冲，直到下游系统请求更多的元素。

5.1. Example

5.1.例子

Let us now demonstrate the example of the create() method:

现在让我们演示一下create()方法的例子。

public class CharacterCreator {
    public Consumer<List<Character>> consumer;

    public Flux<Character> createCharacterSequence() {
        return Flux.create(sink -> CharacterCreator.this.consumer = items -> items.forEach(sink::next));
    }
}

We can notice that the create operator asks us for a FluxSink instead of a SynchronousSink as used in the generate(). In this case, we’ll call next() for every item we have in the list of items, emitting each one by one.

我们可以注意到，create操作符要求我们提供一个FluxSink，而不是SynchronousSink，因为在generate()中使用。在这种情况下，我们将为items列表中的每个项目调用next()，逐一排放。

Let us now use the CharacterCreator with two sequences of characters:

现在让我们用两个字符序列使用CharacterCreator。

@Test
public void whenCreatingCharactersWithMultipleThreads_thenSequenceIsProducedAsynchronously() throws InterruptedException {
    CharacterGenerator characterGenerator = new CharacterGenerator();
    List<Character> sequence1 = characterGenerator.generateCharacters().take(3).collectList().block();
    List<Character> sequence2 = characterGenerator.generateCharacters().take(2).collectList().block();
}

We’ve created two sequences in the above code snippet – sequence1 and sequence2. These sequences serve as the source of character items. Note that we are using the CharacterGenerator instance to get the sequence of characters.

我们在上面的代码片段中创建了两个序列–sequence1和sequence2。这些序列作为字符项的来源。注意，我们正在使用CharacterGenerator实例来获得字符序列。

Let us now define an instance of characterCreator and two thread instances:

现在让我们定义一个characterCreator的实例和两个线程实例。

CharacterCreator characterCreator = new CharacterCreator();
Thread producerThread1 = new Thread(() -> characterCreator.consumer.accept(sequence1));
Thread producerThread2 = new Thread(() -> characterCreator.consumer.accept(sequence2));

We are now creating two thread instances that will provide elements to the publisher. The character element starts flowing into the sequence source when the accept operator is invoked. Next, we subscribe to the new consolidated sequence:

我们现在正在创建两个线程实例，它们将向发布者提供元素。当调用接受操作符时，字符元素开始流向序列源。接下来，我们订阅到新的合并序列。

List<Character> consolidated = new ArrayList<>();
characterCreator.createCharacterSequence().subscribe(consolidated::add);

Notice that createCharacterSequence returns a Flux to which we subscribed and consumes the elements in the consolidated list. Next, let us trigger the whole process that sees items moving on two different threads:

注意，createCharacterSequence返回一个我们订阅的Flux，并消耗consolidated列表中的元素。接下来，让我们触发整个过程，看到项目在两个不同的线程上移动。

producerThread1.start();
producerThread2.start();
producerThread1.join();
producerThread2.join();

Finally, let us verify the operation’s result:

最后，让我们验证一下操作的结果。

assertThat(consolidated).containsExactlyInAnyOrder('a', 'b', 'c', 'a', 'b');

The first three characters in the received sequence come from sequence1. And the last two characters are from sequence2. Since this is an asynchronous operation, the order of elements from those sequences isn’t guaranteed.

收到的序列中的前三个字符来自序列1。而最后两个字符来自序列2。由于这是一个异步操作，来自这些序列的元素的顺序不被保证。

6. Flux Create vs. Flux Generate

6.Flux Create vs. Flux Generate

Following are a few differences between the create and generate operations:

以下是创建和生成操作之间的一些区别。

Flux Create	Flux Generate
This method accepts an instance of Consumer<FluxSink>	This method accepts an instance of Consumer<SynchronousSink>
Create method calls the consumer only once	Generate method calls the consumer method multiple times based on the need of the downstream application
A consumer can emit 0..N elements immediately	Can emit only one element
The publisher is unaware of the downstream state. Therefore create accepts an Overflow strategy as an additional parameter for flow control	The publisher produces elements based on the downstream application need
The FluxSink lets us emit elements using multiple threads if required	Not useful for multiple threads as it emits only one element at a time

7. Conclusion

7.结语

In this article, we discussed the differences between the create and generate methods of Flux API.

在这篇文章中，我们讨论了Flux API的创建和生成方法之间的区别。

First, we introduced the notion of reactive programming and talked about the Flux API. We then discussed the create and generate methods of Flux API. Finally, we provided a list of differences between the create and generate methods of the Flux API.

首先，我们介绍了反应式编程的概念并谈到了Flux API。然后我们讨论了Flux API的创建和生成方法。最后，我们提供了一个Flux API的创建和生成方法之间的差异列表。

The source code for this tutorial is available over on GitHub.

本教程的源代码可在GitHub上获得，。