Ich habe viele Antworten zu RxJava gefunden, möchte aber verstehen, wie es in Reactor funktioniert.

Mein aktuelles Verständnis ist sehr vage, ich neige dazu, mir Map als synchron und flatMap als asynchron vorzustellen, aber ich komme damit nicht wirklich klar.

Hier ist ein Beispiel:

files.flatMap { it ->
    Mono.just(Paths.get(UPLOAD_ROOT, it.filename()).toFile())
        .map {destFile ->
            destFile.createNewFile()
            destFile    
        }               
        .flatMap(it::transferTo)
}.then()  

Ich habe Dateien (a Flux<FilePart>) und ich möchte es auf einige kopieren UPLOAD_ROOT auf dem Server.

Dieses Beispiel stammt aus einem Buch.

Ich kann alles ändern .map Zu .flatMap und umgekehrt und alles funktioniert immer noch. Ich frage mich, was der Unterschied ist.

• map ist für synchrone, nicht blockierende 1-zu-1-Transformationen
• flatMap ist für asynchrone (nicht blockierende) 1-zu-N-Transformationen

Der Unterschied ist in der Methodensignatur sichtbar:

• map nimmt ein Function<T, U> und gibt a zurück Flux<U>
• flatMap nimmt ein Function<T, Publisher<V>> und gibt a zurück Flux<V>

Das ist der wichtigste Hinweis: Sie dürfen pass a Function<T, Publisher<V>> zu einem mapaber es würde nicht wissen, was es damit machen soll Publishersund das würde zu einem führen Flux<Publisher<V>>eine Folge inaktiver Herausgeber.

Andererseits, flatMap erwartet a Publisher<V> für jede T. Es weiß, was es damit machen soll: es abonnieren und seine Elemente in der Ausgabesequenz weitergeben. Daraus ergibt sich der Rückgabetyp Flux<V>: flatMap wird jedes Innere abflachen Publisher<V> in die Ausgabesequenz von alle Die VS.

Zum 1-N-Aspekt:

für jede <T> Eingabeelement, flatMap ordnet es einem zu Publisher<V>. In einigen Fällen (z. B. bei einer HTTP-Anfrage) gibt dieser Herausgeber nur ein Element aus. In diesem Fall sind wir einer Asynchronität ziemlich nahe map.

Aber das ist der degenerierte Fall. Der generische Fall ist, dass a Publisher kann mehrere Elemente emittieren und flatMap funktioniert genauso gut.

Stellen Sie sich zum Beispiel vor, Sie haben eine reaktive Datenbank und erstellen eine FlatMap aus einer Folge von Benutzer-IDs mit einer Anfrage, die den Satz von Benutzer-IDs zurückgibt Badge. Am Ende erhältst du eine Single Flux<Badge> aller Abzeichen aller dieser Benutzer.

Ist map wirklich synchron und nicht blockierend?

Ja: Sie ist synchron in der Art und Weise, wie der Operator sie anwendet (ein einfacher Methodenaufruf, und dann gibt der Operator das Ergebnis aus) und nicht blockierend in dem Sinne, dass die Funktion selbst den Operator, der sie aufruft, nicht blockieren sollte. Mit anderen Worten, es sollte keine Latenz verursachen. Das liegt daran, dass a Flux ist insgesamt immer noch asynchron. Wenn es mitten in der Sequenz blockiert, wirkt es sich auf den Rest aus Flux Verarbeitung oder sogar andere Flux.

Wenn Ihre Kartenfunktion blockiert/Latenz verursacht, aber nicht in die Rückgabe einer konvertiert werden kann Publisherhalten publishOn/subscribeOn um diese Blockierungsarbeit in einem separaten Thread auszugleichen.

Die FlatMap-Methode ähnelt der Map-Methode mit dem entscheidenden Unterschied, dass der Lieferant, den Sie ihr bereitstellen, eine zurückgeben sollte Mono<T> oder Flux<T>.

Die Verwendung der Kartenmethode würde zu einem führen Mono<Mono<T>>
wohingegen die Verwendung von flatMap zu einem führt Mono<T>.

Dies ist beispielsweise nützlich, wenn Sie zum Abrufen von Daten einen Netzwerkaufruf mit einer Java-API durchführen müssen, die ein Mono zurückgibt, und dann einen weiteren Netzwerkaufruf, der das Ergebnis des ersten benötigt.

// Signature of the HttpClient.get method
Mono<JsonObject> get(String url);

// The two urls to call
String firstUserUrl = "my-api/first-user";
String userDetailsUrl = "my-api/users/details/"; // needs the id at the end

// Example with map
Mono<Mono<JsonObject>> result = HttpClient.get(firstUserUrl).
  map(user -> HttpClient.get(userDetailsUrl + user.getId()));
// This results with a Mono<Mono<...>> because HttpClient.get(...)
// returns a Mono

// Same example with flatMap
Mono<JsonObject> bestResult = HttpClient.get(firstUserUrl).
  flatMap(user -> HttpClient.get(userDetailsUrl + user.getId()));
// Now the result has the type we expected

Außerdem ermöglicht es eine präzise Fehlerbehandlung:

public UserApi {
  
  private HttpClient httpClient;
    
  Mono<User> findUser(String username) {
    String queryUrl = "http://my-api-address/users/" + username;
    
    return Mono.fromCallable(() -> httpClient.get(queryUrl)).
      flatMap(response -> {
        if (response.statusCode == 404) return Mono.error(new NotFoundException("User " + username + " not found"));
        else if (response.statusCode == 500) return Mono.error(new InternalServerErrorException());
        else if (response.statusCode != 200) return Mono.error(new Exception("Unknown error calling my-api"));
        return Mono.just(response.data);
      });
  }
                                           
}

Wie die Karte intern im Reactor funktioniert.

Ein … Erstellen Player Klasse.

@Data
@AllArgsConstructor
public class Player {
        String name;
        String name;
}

Erstellen Sie nun einige Instanzen von Player Klasse

Flux<Player> players = Flux.just(
        "Zahid Khan",
        "Arif Khan",
        "Obaid Sheikh")
        .map(fullname -> {
            String[] split = fullname.split("\\s");
            return new Player(split[0], split[1]);
        });

StepVerifier.create(players)
          .expectNext(new Player("Zahid", "Khan"))
          .expectNext(new Player("Arif", "Khan"))
          .expectNext(new Player("Obaid", "Sheikh"))
          .verifyComplete();

Es ist wichtig, über map() zu verstehen, dass die Zuordnung synchron durchgeführt wird, da jedes Element vom Quell-Flux veröffentlicht wird. Wenn Sie das Mapping asynchron durchführen möchten, sollten Sie die Operation flatMap() in Betracht ziehen.

Wie FlatMap intern funktioniert.

Flux<Player> players = Flux.just(
      "Zahid Khan", 
      "Arif Khan", 
      "Obaid Sheikh")
      .flatMap(
            fullname -> 
                  Mono.just(fullname).map(p -> {
                        String[] split = p.split("\\s");
                        return new Player(split[0], split[1]);
        }).subscribeOn(Scheduler.parallel()));

        List<Player> playerList = Arrays.asList(
                  new Player("Zahid", "Khan"),
                  new Player("Arif", "Khan"), 
                  new Player("Obaid", "Sheikh"));

        StepVerifier.create(players).expectNextMatches(player ->         
                playerList.contains(player))    
                        .expectNextMatches(player ->  
                                playerList.contains(player))
                        .expectNextMatches(player -> 
                                playerList.contains(player))
                        .expectNextMatches(player -> 
                                playerList.contains(player))
                        .verifyComplete();

Intern wird in einer Flatmap() eine Map()-Operation für Mono ausgeführt, um den String in Player umzuwandeln. Außerdem, subcribeOn() gibt an, dass jedes Abonnement in einem parallelen Thread erfolgen soll. In Abwesenheit von subscribeOn() fungiert flatmap() als synchronisiert.

Die Karte ist für synchrone, nicht blockierende Eins-zu-Eins-Transformationen vorgesehen, während die FlatMap für asynchrone (nicht blockierende) Eins-zu-Viele-Transformationen vorgesehen ist.