Flux.create与Flux.generate之间的区别

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:

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

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

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:

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

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

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.

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

This method accepts an instance of Consumer

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.

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.

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