Comment implémenter le streaming réactif à l'aide de l'API Flow dans Java 9 ?

Flow API est le support officiel de la spécification des flux réactifs depuis Java 9. C'est une combinaison de modèles Iterator et Observer . Flow API est une spécification d'interopérabilité, pas une API d'utilisateur final comme RxJava.

Flow API se compose de quatre interfaces de base :

• Abonné : L'abonné s'abonne aux rappels auprès de l'éditeur.
• Publisher : Publisher publie un flux d'éléments de données aux abonnés enregistrés.
• Abonnements : Le lien entre éditeurs et abonnés.
• Processeur : Le processeur se situe entre l'éditeur et l'abonné et transforme un flux en un autre.

Dans l'exemple ci-dessous, nous créons un abonné de base qui demande un objet de données, l'imprime et demande un autre objet de données. Nous pouvons utiliser l'implémentation de l'éditeur fournie par Java (SubmissionPublisher) pour terminer notre session.

Exemple

import java.util.concurrent.Flow;
import java.util.List;
import java.util.concurrent.SubmissionPublisher;

class MySubscriber<T>implements <strong>Flow.Subscriber<T></strong> {
   private <strong>Flow.Subscription</strong> subscription;
   <strong>@Override</strong>
   public void <strong>onSubscribe</strong>(Flow.Subscription subscription) {
      this.subscription = subscription;
      this.subscription.request(1);
   }
   <strong>@Override</strong>
   public void <strong>onNext</strong>(T item) {
      System.out.println(item);
      subscription.request(1);
   }
   <strong>@Override</strong>
   public void <strong>onError</strong>(Throwable throwable) {
      throwable.printStackTrace();
   }
   <strong>@Override</strong>
   public void <strong>onComplete</strong>() {
      System.out.println("Done");
   }
}

<strong>// main class</strong>
public class FlowTest {
   public static void main(String args[]) {
      <strong>List<String></strong> items = <strong>List.of</strong>("1", "2", "3", "4", "5", "6", "7", "8", "9", "10");
      <strong>SubmissionPublisher<String></strong> publisher = new SubmissionPublisher<>();
      publisher.<strong>subscribe</strong>(new MySubscriber<>());
      items.forEach(s -> {
         try {
            Thread.sleep(1000);
         } catch (InterruptedException e) {
            e.printStackTrace();
         }
         publisher.<strong>submit</strong>(s);
      });
      publisher.close();
   }
}

Sortie

<strong>1
2
3
4
5
6
7
8
9</strong>
<strong>10
Done</strong>

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!