Triển khai interface Subscriber trong Java 9

Java 9 hỗ trợ tạo Reactive Streams bằng cách giới thiệu một vài interface: Publisher, Subscriber, Subscription, và lớp SubmissionPublisher triển khai interface Publisher. Mỗi interface đóng vai trò khác nhau tương ứng với nguyên tắc của Reactive Streams.

Chúng ta có thể sử dụng interface Subscriber để đăng ký dữ liệu được xuất bản bởi một publisher. Cần triển khai interface Subscriber và cung cấp cài đặt cho các phương thức trừu tượng.

Các phương thức của interface Flow.Subscriber:

• onComplete(): Được gọi khi đối tượng Publisher hoàn thành vai trò.
• onError(): Được gọi khi có lỗi xảy ra trong Publisher và được thông báo cho Subscriber.
• onNext(): Được gọi bất cứ khi Publisher có thông tin mới cần thông báo cho tất cả Subscriber.
• onSubscribe(): Được gọi khi Publisher thêm Subscriber.

Ví dụ

import java.util.concurrent.Flow; 
import java.util.concurrent.SubmissionPublisher; 
import java.util.stream.IntStream; 
public class SubscriberImplTest { 
  public static class Subscriber implements Flow.Subscriber<Integer> { 
    private Flow.Subscription subscription; 
    private boolean isDone; 
    @Override 
    public void onSubscribe(Flow.Subscription subscription) { 
      System.out.println("Subscribed"); 
      this.subscription = subscription; 
      this.subscription.request(1); 
    } 
    @Override 
    public void onNext(Integer item) { 
      System.out.println("Processing " + item); 
      this.subscription.request(1); 
    } 
    @Override 
    public void onError(Throwable throwable) { 
      throwable.printStackTrace(); 
    } 
    @Override 
    public void onComplete() { 
      System.out.println("Processing done"); 
      isDone = true; 
    } 
  } 
  public static void main(String args[]) throws InterruptedException { 
    SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>(); 
    Subscriber subscriber = new Subscriber(); 
    publisher.subscribe(subscriber); 
    IntStream intData = IntStream.rangeClosed(1, 10); 
    intData.forEach(publisher::submit); 
    publisher.close(); 
    while(!subscriber.isDone) { 
      Thread.sleep(10); 
    } 
    System.out.println("Done"); 
  } 
}

Kết quả

Subscribed Processing 1 Processing 2 Processing 3 Processing 4 Processing 5 Processing 6 Processing 7 Processing 8 Processing 9 Processing 10 Processing done Done