Java
Özet - Reactive
Flow API ve SubmissionPublisher'ı birleştirerek bir sıcaklık izleyici yazarsın.
Özet - Reactive
Bu grupta Reactive Streams'in temellerini gördük: Flow API ile üç arayüz, Publisher ve Subscriber etkileşimi, SubmissionPublisher ile hazır Publisher, backpressure ile talep denetimi ve Project Reactor Mono ile bildirimsel akış. Bunlar asenkron veri akışını düzenler.
Flow API sözleşmesi standarttır; Reactor bunu bildirimsel bir üslupla sarmalar. Backpressure, talep temelli iletimle bellek şişmesini önler. Şimdi iki küçük görevle akışı ve aboneliği tekrar edeceğiz.
Sözleşme ve kütüphane
Flow API yalnızca sözleşmeyi tanımlar; nasıl iletileceği uygulamaya bağlıdır. SubmissionPublisher asenkron bir uygulama sunarken, kendi senkron Publisher'ını da yazabilirsin. Reactor ise bu sözleşmenin üzerine bildirimsel işlemler ekler.
Önemli olan, hangi yolu seçersen seç, akışın Publisher'dan Subscriber'a olması ve talebin Subscriber'dan gelmesidir. Bu iki ilke, tüm reactive uygulamaların temelidir. Akış tamamlandığında onComplete, hata olduğunda onError gelir.
Uygulama Mono ile süz ve dönüştür
Mono.just(3) ile bir Mono oluştur. .filter(x -> x > 1) ile süz. .map(x -> x + 1) ile bir artır. .subscribe(System.out::println) ile sonucu yazdır.
Örnek çözümü göster
import reactor.core.publisher.Mono;
public class MonoFilterDemo {
public static void main(String[] args) {
Mono.just(3)
.filter(x -> x > 1)
.map(x -> x + 1)
.subscribe(System.out::println);
}
}
Uygulama İki aboneli say
Bir EmptySubscriber sınıfı tanımla: Subscriber<Integer> arayüzünü uygulasın (onSubscribe request(1) yapsın, diğerleri boş). main içinde new SubmissionPublisher<Integer>() oluştur, iki EmptySubscriber abone ekle ve getNumberOfSubscribers sonucunu System.out.println ile yazdır. Sonra pub.close() ile kapat.
Örnek çözümü göster
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.Flow.*;
class EmptySubscriber implements Subscriber<Integer> {
public void onSubscribe(Subscription s) { s.request(1); }
public void onNext(Integer v) { }
public void onError(Throwable t) { }
public void onComplete() { }
}
public class SubCountDemo {
public static void main(String[] args) {
SubmissionPublisher<Integer> pub = new SubmissionPublisher<>();
pub.subscribe(new EmptySubscriber());
pub.subscribe(new EmptySubscriber());
System.out.println(pub.getNumberOfSubscribers());
pub.close();
}
}
Kendini test et
-
Backpressure hangi taraf tarafından belirlenir?
- Subscriber
- Channel
- Publisher
- Buffer
Cevabı göster
Subscriber — Subscriber, request ile talebi belirler.
-
subscribe çağrısını tamamla
pub.____(new EmptySubscriber());
Cevabı göster
pub.subscribe(new EmptySubscriber());
Sıradaki ders: native metot
