Cara menggunakan ciri baharu API pengaturcaraan responsif java9 Reactive Stream

1. Java9 Reactive Stream API

Java 9 menyediakan satu set antara muka untuk mentakrifkan pengaturcaraan aliran reaktif. Semua antara muka ini ditakrifkan dalam kelas java.util.concurrent.Flow sebagai antara muka dalaman statik.

Berikut ialah beberapa peranan dan konsep penting dalam pengaturcaraan reaktif Java Mari kita fahami secara ringkas dahulu

Penerbit berpotensi tidak terhad dalam bilangan Pengeluar elemen data tertib. . Ia menerbitkan beberapa elemen data tertentu kepada pelanggan semasa berdasarkan permintaan yang diterima (langganan).

Pelanggan melanggan dan menerima elemen data daripada penerbit. Selepas mewujudkan hubungan langganan dengan penerbit, penerbit menghantar token langganan (langganan) kepada pelanggan, dan pelanggan boleh meminta bilangan elemen data yang diterbitkan oleh penerbit mengikut keupayaan pemprosesannya sendiri.

Token langganan (langganan) mewakili hubungan langganan yang diwujudkan antara pelanggan dan penerbit. Apabila perhubungan langganan diwujudkan, penerbit menyerahkannya kepada pelanggan. Pelanggan menggunakan token langganan untuk berinteraksi dengan penerbit, seperti meminta bilangan elemen data atau berhenti melanggan.

2. Empat antara muka utama pengaturcaraan reaktif Java

2.1. Antara Muka Pelanggan (Antara Muka Langganan Pelanggan)

public static interface Subscriber<T> {
    public void onSubscribe(Subscription subscription);
    public void onNext(T item);
    public void onError(Throwable throwable);
    public void onComplete();
}

onSubscribe: Terima langganan daripada pelanggan di penerbit Dipanggil selepas tindakan itu tetapi sebelum menerbitkan sebarang mesej langganan. Objek token langganan Subscription yang baru dibuat dihantar kepada pelanggan melalui kaedah ini.

onNext: Fungsi pemprosesan item data seterusnya yang akan diproses

onError: Dipanggil apabila penerbit atau langganan mengalami ralat yang tidak boleh dipulihkan

onComplete : Dipanggil apabila tiada panggilan pelanggan (termasuk kaedah onNext()) berlaku.

2.2.Antara Muka Langganan (Antara Muka Token Langganan)

Objek token langganan diluluskan melalui kaedah Subscriber.onSubscribe()

public static interface Subscription {    public void request(long n);    public void cancel();}

request(long n) berada di sebalik konsep bukan- menyekat tekanan belakang Kaedah utama. Pelanggan menggunakannya untuk meminta lebih daripada n item penggunaan. Dengan cara ini, pelanggan mengawal jumlah data yang boleh diterimanya pada masa ini. cancel()Pelanggan harus mengambil inisiatif untuk membatalkan langganan mereka Selepas pembatalan, mereka tidak akan menerima sebarang mesej data.

2.3.Antara Muka Penerbit

@FunctionalInterface
public static interface Publisher<T> {
    public void subscribe(Subscriber<? super T> subscriber);
}

Panggil kaedah ini untuk mewujudkan hubungan langganan mesej antara Pelanggan dan Penerbit.

2.4. Antara Muka Pemproses

Pemproses boleh bertindak sebagai pelanggan dan penerbit, dan memainkan peranan untuk menukar elemen dalam saluran paip penerbit-pelanggan. Digunakan untuk menerima dan menukar elemen data jenis T daripada penerbit kepada data jenis R dan menerbitkannya.

public static interface Processor<T,R> extends Subscriber<T>, Publisher<R> {
}

2. Kes praktikal

Sekarang kita perlu melaksanakan empat antara muka di atas untuk melengkapkan pengaturcaraan reaktif

Subscription InterfaceAntara muka token langganan biasanya tidak memerlukan kita melakukannya diri kita Untuk melaksanakannya melalui pengaturcaraan, kita hanya perlu mengetahui maksud kaedah request() dan kaedah cancel().

Publisher InterfaceAntara muka penerbit, Java 9 telah menyediakan kami dengan pelaksanaan SubmissionPublisher secara lalai Selain kaedah melaksanakan antara muka Penerbit, kelas pelaksanaan ini menyediakan kaedah yang dipanggil submit() untuk melengkapkan penghantaran. daripada data mesej.

Subscriber Interface Antara muka pelanggan biasanya perlu dilaksanakan oleh kami sendiri. Kerana selepas langganan data diterima, perniagaan yang berbeza mempunyai logik pemprosesan yang berbeza.

ProcessorIa sebenarnya adalah koleksi Antara Muka Penerbit dan Antara Muka Pelanggan Antara muka ini perlu dilaksanakan jika penukaran jenis data dan pemprosesan data diperlukan

Contoh berikut melaksanakan ungkapan rentetan Data. pemprosesan langganan mesej

melaksanakan Antara Muka Pelanggan

import java.util.concurrent.Flow;
public class MySubscriber implements Flow.Subscriber<String> {
  private Flow.Subscription subscription;  //订阅令牌
  @Override
  public void onSubscribe(Flow.Subscription subscription) {
      System.out.println("订阅关系建立onSubscribe: " + subscription);
      this.subscription = subscription;
      subscription.request(2);
  }
  @Override
  public void onNext(String item) {
      System.out.println("item: " + item);
      // 一个消息处理完成之后，可以继续调用subscription.request(n);向发布者要求数据发送
      //subscription.request(n);
  }
  @Override
  public void onError(Throwable throwable) {
      System.out.println("onError: " + throwable);
  }
  @Override
  public void onComplete() {
      System.out.println("onComplete");
  }
}

Penerbit mesej SubmissionPublisher

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
public class SubmissionPublisherExample {
  public static void main(String[] args) throws InterruptedException {
      ExecutorService executor = Executors.newFixedThreadPool(1);
      SubmissionPublisher<String> sb = new SubmissionPublisher<>(executor, Flow.defaultBufferSize());
      sb.subscribe(new MySubscriber());   //建立订阅关系，可以有多个订阅者
      sb.submit("数据 1");  //发送消息1
      sb.submit("数据 2"); //发送消息2
      sb.submit("数据 3"); //发送消息3
      executor.shutdown();
  }
}

Hasil cetakan konsol

Hubungan langganan Cipta
onSubscribe .util.concurrent.SubmissionPublisher$BufferedSubscription@27e81a39
item: Data 1
item: Data 2

Sila ambil perhatian: Walaupun penerbit menyerahkan 3 item Data, MySubscriber hanya menerima 2 keping data untuk diproses. Ini kerana kami menggunakan MySubscriber#onSubscribe() dalam kaedah subscription.request(2);. Ini ialah kesan pengaturcaraan reaktif "tekanan belakang". Sebanyak data yang saya ada untuk memproses, saya akan memberitahu penerbit mesej berapa banyak data yang perlu diberikan.

Atas ialah kandungan terperinci Cara menggunakan ciri baharu API pengaturcaraan responsif java9 Reactive Stream. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!