Java 9 hỗ trợ tạo Luồng phản ứng bằng cách giới thiệu một vài giao diện: Publisher , Người đăng ký , Đăng ký và SubmissionPublisher lớp triển khai Publisher giao diện. Mỗi giao diện có thể đóng một vai trò khác nhau tương ứng với các nguyên tắc của Phản ứng Luồng .
Chúng tôi có thể sử dụng Người đăng ký giao diện để đăng ký dữ liệu được xuất bản bởi nhà xuất bản . Chúng tôi cần triển khai Người đăng ký giao diện và cung cấp một triển khai cho các phương thức trừu tượng.
Phương thức giao diện Flow.Subscriber:
- onComplete (): Phương thức này đã được gọi khi đối tượng Publisher hoàn thành vai trò của nó.
- onError (): Phương thức này đã được gọi khi có sự cố xảy ra trong Publisher và được thông báo cho Người đăng ký.
- onNext (): Phương thức này được gọi bất cứ khi nào Nhà xuất bản có thông tin mới cần được thông báo cho tất cả Người đăng ký.
- onSubscribe (): Phương thức này đã được gọi khi Nhà xuất bản thêm Người đăng ký.
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");
}
} Đầu ra
Subscribed Processing 1 Processing 2 Processing 3 Processing 4 Processing 5 Processing 6 Processing 7 Processing 8 Processing 9 Processing 10 Processing done Done