Comment implémenter le streaming réactif à l'aide de l'API Flow dans Java 9 ?-javaDidacticiel-php.cn

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

王林

Libérer： 2023-09-19 23:49:02

avant

824 Les gens l'ont consulté

如何在Java 9中使用Flow API实现响应式流？

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

Copier après la connexion

Sortie

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

Copier après la connexion

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!