Difference Between Flux.create and Flux.generate – Flux.create和Flux.generate之间的区别

最后修改: 2022年 6月 20日

1. Introduction


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


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 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.


3. Maven Dependencies


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



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


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();
            if (value == 'z') {
            return state + 1;

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


  • 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.


Let’s verify the generated sequence with StepVerifier:


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

      .expectNext('a', 'b', 'c')

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.


5. Flux Create


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.


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


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


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.


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


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.


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


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<>();

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:



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.


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


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.