2. Bir giriş örneği

RxJava ile ilk tanışmam, internette bulduğum dersler ve öğreticiler aracılığıyla oldu. Teoride alışık olmadığım ve anlamakta zorlandığım kavramlar kullanılıyor olmasının yanı sıra, en önemlisi bunun gerçek hayatta ne işe yarayacağını hiç anlayamıyordum. Bu nedenle, RxJava'in kullanımıyla kod yazımının gerçekten basitleştiği (umarım basit bir) örnekle başlayacağız ve buradan yola çıkarak bu kütüphanenin önemli unsurlarını belirlemeye çalışacağız.

RxJava kütüphanesi şu konsepte dayanır: T türündeki Observable<T> öğelerinden oluşan bir akış, bir veya daha fazla Subscriber<T> (abone, gözlemci, tüketici) tarafından gözlemlenir. RxJava kütüphanesi, Observable<T> akışının bir T1 iş parçacığında, gözlemcisi Subscriber<T>'nin ise bir T2 iş parçacığında çalışmasına olanak tanır; böylece geliştiricininbu iş parçacıklarının yaşam döngüsünü yönetmekle veya iş parçacıkları arasında veri paylaşımı ve genel bir görevi yürütmek için iş parçacıklarının senkronizasyonu gibi doğal olarak zor sorunlarla uğraşmasına gerek kalmadan. Böylece asenkron programlamayı kolaylaştırır.

Bir Observable<T> akışı, T türündeki öğeleri üretir ve bu öğeler üretildikçe gözlemlenebilir hale gelir. Gözlemci ve gözlemlenebilir (konuşma dilinde Observable<T> türünü ifade etmek için kullanılır) aynı iş parçacığında bulunuyorsa, gözlemlenebilir, gözlemci i öğesini tükettiğinde ancak (i+1) öğesini üretebilir. Bu mimarinin faydalı olduğu durumlar çok azdır. Gözlemci ve gözlemlenebilir aynı iş parçacığında değilse, gözlemlenebilir ve gözlemcisi birbirinden bağımsız davranır: gözlemlenebilir kendi hızında üretir, gözlemci ise kendi hızında tüketir. Kütüphanenin önemi işte burada yatmaktadır. Şimdiye kadar hep tek bir gözlemciden bahsettik. Oysa gerçekte, bir gözlemlenebilirin istediği kadar gözlemcisi olabilir.

2.1. Örnek uygulamanın mimarisi

Örnek uygulamanın mimarisi şu şekildedir:

• [1]'te bir hizmet katmanı rastgele sayı listeleri sağlar. Bu katman, onu kullanan [swing] yöntemi ile aynı iş parçacığında çalıştırılır. Böylece sayıları senkron bir şekilde sağlar;
• [2]'te, RxJava ile uygulanan ince bir uyarlama katmanı, [swing] katmanına aynı hizmetin asenkron bir uygulamasını sunmayı sağlar: bu, onu kullanan [swing] yönteminden farklı bir iş parçacığında çalıştırılabilir;
• [4] çağrısı senkron iken, [5-6] çağrısı asenkron;

Burada göstermek istediğimiz şey, Rx kütüphanesinin senkron bir arayüzü asenkron bir arayüze kolayca dönüştürmeye olanak tanıdığıdır. Bu neden yararlıdır? Bir Swing arayüzündeki olaylar, genellikle olay döngüsü (event loop) olarak adlandırılan bir iş parçacığında işlenir. Olaylar bir kuyrukta beklemeye alınır ve sırayla işlenir. Ei+1 olayı, ancak önceki Ei olayı tamamen işlendiğinde işlenebilir. Bu nedenle, grafik arayüzün duyarlı kalması için bir olayın işlenmesinin mümkün olduğunca kısa sürmesi önemlidir. Bazen bir olayın işlenmesi çok zaman alabilir. Bu durum, söz konusu işlem ağ erişimlerini içeriyorsa geçerlidir. Kullanıcı için kabul edilemez bir şekilde grafik arayüzün donmasını istemiyorsak, bu ağ erişimlerinin olay döngüsünden ayrı iş parçacıklarında gerçekleştirilmesi ve böylece olay döngüsünün boşaltılması gerekir. Bu noktada, haklı olarak zor kabul edilen eşzamanlı programlama alanına gireriz (birden fazla iş parçacığı paralel olarak çalışır). Rx kütüphanesi, bu soruna basit ve zarif bir çözüm sunar.

Uzun süren işlemleri simüle etmek için, örnekteki hizmet, grafik arayüzün davranışını görebilmemiz amacıyla belirli bir bekleme süresinden sonra rastgele sayılarını verir.

2.2. L'exécutable

Örnek uygulamanın yürütülebilir dosyası, örnekler klasöründeki [dvp/executables] klasöründe bulunur:

[swing-01] dosyasını çalıştırmak için, çalıştırma işleminin yapıldığı bilgisayarın yapılandırmasına bağlı olarak çeşitli yöntemler mevcuttur. Örneğin, [1-3] adımlarını takip edebilirsiniz. Böylece aşağıdaki grafik kullanıcı arayüzü karşınıza çıkar:

• Arayüzde iki sekme bulunur: [1-2]; bunlardan biri [Request] rastgele sayı üretme hizmetine yapılan istek için, diğeri ise [Response] alınan sayıların görüntülenmesi içindir;
• [3] sekmesinde, servise kaç adet istek gönderileceği belirtilir;
• [4]'te, istenen sayıların üretileceği aralık [a,b] belirtilir;
• [5]'te, hizmet tarafından döndürülecek değer sayısı, kullanıcı tarafından belirlenen [minCount, maxCount] aralığı içindeki rastgele bir sayı olacaktır;
• [6]'te, yanıtını göndermeden önce, hizmet, delay milisaniye bekleyecektir; burada delay, kullanıcı tarafından belirlenen [minDelay, maxDelay] aralığında rastgele bir sayıdır;
• varsayılan olarak, [swing] katmanı hizmetin senkron arayüzüne yönelir. Asenkron katmana yönelmek için kullanıcı [7] seçeneğini işaretleyecektir. Bu durumda, oluşturma hizmeti grafik arayüzün olay döngüsünden ayrı iş parçacıklarında çalışacaktır. Rx kütüphanesi, bu iş parçacıklarını oluşturmak için çeşitli stratejilere sahiptir. Kullanıcı, [8] seçeneğinden istediği stratejiyi seçebilir;
• sayıların oluşturulması [9] düğmesi ile gerçekleştirilir;

• [10] ile sonuçların görüntülenmesi sağlanır. Bu sonuçların yapısını açıklayacağız;
• [11]'te, elde edilen sonuçların sayısı;
• [12]'te, milisaniye cinsinden yürütme süresi;
• [13]'te, kullanıcı yürütmeyi iptal etme seçeneğine sahiptir;

Her sonuç şu biçimdedir:

{"idClient":0,"serviceResponse":{"delay":412,"aleas":[146,115,128,174,159,112,162,127],"executedOn":"RxComputationThreadPool-6"},"observedOn":"AWT-EventQueue-0","requestAt":"02:42:47:708","responseAt":"02:42:52:931"}

• [idClient]: isteğin numarası. Oluşturma hizmetine birden fazla istek gönderildiğini hatırlatmak isteriz;
• [delay]: hizmetin sonucunu göndermeden önce beklediği süre (milisaniye cinsinden);
• [aleas]: hizmet tarafından döndürülen rastgele sayılar;
• [executedOn]: hizmetin çalıştığı iş parçacığının adı;
• [observedOn]: Sonucu görüntüleyen iş parçacığının adı. Bir Swing arayüzünde bu, yalnızca olay döngüsünün iş parçacığı olabilir; buradaki ad [AWT-EventQueue-0]'tir;
• [requestAt]: [heures:minutes:secondes:millisecondes] biçiminde isteğin saati;
• [responseAt]: sonuçların alındığı saat, aynı biçimde;

Şimdi, örneği anlamak için yararlı olan kod bölümlerini sunacağız.

2.3. Senkron arayüz

[1] hizmet katmanı aşağıdaki arayüzü sunar:

public interface IService {
  // [a,b]'te rastgele sayılar
  // n aralığında n rastgele sayı üretilir [minCount, maxCount]
  // sayılar, delay milisaniye bekleme süresinden sonra üretilir,
  // burada [delay], [minDelay, maxDelay] aralığında bir rastgele sayıdır
  public ServiceResponse getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay);
}

[ServiceResponse] yanıtı şöyledir:

public class ServiceResponse {

  // hizmetin bekleme süresi
  private int delay;
  // rastgele sayılar
  private List<Integer> aleas;
  // çalıştırma iş parçacığı
  private String executedOn;

  // oluşturucular

  public ServiceResponse(int delay, List<Integer> aleas) {
    executedOn = Thread.currentThread().getName();
    this.delay = delay;
    this.aleas = aleas;
  }

  // alıcı ve ayarlayıcılar
...
}

Yanıt üç öğeden oluşur:

• 6. satır: üretilen rastgele sayılar;
• 4. satır: hizmetin sonucunu vermeden önce beklediği süre;
• 8. satır: hizmetin yürütme iş parçacığı;

2.4. Senkron çağrı

Şimdi, [swing] katmanının [1] hizmetine yaptığı [4] senkron çağrısını ayrıntılı olarak inceleyeceğiz:

  private void doGenerateWithService() {
    // bekleme başlangıcı
    beginWaiting();
    try {
      for (int i = 0; i < nbRequests; i++) {
        UiResponse uiResponse = new UiResponse();
        uiResponse.setIdClient(i);
        uiResponse.setServiceResponse(service.getAleas(a, b, minCount, maxCount, minDelay, maxDelay));
        uiResponse.setResponseAt();
        model.add(0, jsonMapper.writeValueAsString(uiResponse));
        jLabelNbReponses.setText(String.valueOf(Integer.parseInt(jLabelNbReponses.getText()) + 1));
      }
    } catch (JsonProcessingException | RuntimeException e) {
      System.out.println(e);
    }
    // bekleme sonu
    endWaiting();
}

• 5-12. satırlar: kullanıcı tarafından talep edilen [nbRequests] isteklerinin yürütme döngüsü;
• 8. satır: [service], 2.3. paragrafta tanıtılan senkron [IService] arayüzünün uygulamasıdır;
• 10. satır: [model], [Response] sekmesindeki JList bileşeni tarafından görüntülenen şablondur. Bu şablonun öğeleri, aşağıdaki [UiResponse] türündeki öğelerin jSON dizeleridir:

public class UiResponse {

  // müşteri kimliği
  private int idClient;
  // hizmet yanıtı
  private ServiceResponse serviceResponse;
  // gözlem iş parçacığının adı
  private String observedOn;
  // istek saati
  private String requestAt;
  // yanıt saati
  private String responseAt;

  // yapıcılar

  public UiResponse() {
    observedOn = Thread.currentThread().getName();
    requestAt = getTimeStamp();
  }
  // özel yöntemler

  private String getTimeStamp() {
    return new SimpleDateFormat("hh:mm:ss:SSS").format(Calendar.getInstance().getTime());
  }

  // getter ve setter'lar
...
}

• 6. satır: sayı üretme hizmetinin yanıtı;
• 4. satır: yanıtlanan isteğin numarası;
• 8. satır: bu yanıtın görüntüleme iş parçacığı. Daha önce de belirtildiği gibi, bu her zaman olay döngüsünün iş parçacığı olacaktır;
• 10. ve 12. satırlar: isteğin ve yanıtın zamanı;

2.5. Senkron çağrıların test edilmesi

Aşağıdaki yapılandırmayı çalıştırıyoruz:

[Response] sekmesinde şu sonuçları elde ediyoruz:

• [1-2]'te, talep edildiği gibi 10 yanıt aldık. Bu yanıtlar, geldikleri sıraya göre ilk sıraya yerleştirildi. Bunların istek sırasına göre alındığı görülüyor;
• hepsi [AWT-EventQueue-0] olay döngüsü iş parçacığında yürütülmüş ve görüntülenmiştir. Dolayısıyla istekler bu iş parçacığında birbiri ardına yürütülmüştür. Eşzamanlı istek olmamıştır;
• burada görünmeyen şey, yürütme sırasında grafik arayüzün donmuş olmasıdır. Örneğin, gelen yanıtları görmek için [Response] sekmesine erişmek ya da [Annuler] düğmesiyle yürütmeyi durdurmak mümkün değildir. Bu düğme [Request] sekmesinde olsaydı bile kullanılamazdı. Zira bu durumda iki olay söz konusu olurdu:
  • [Générer] düğmesine tıklama;
  • [Annuler] düğmesine tıklama;

[Annuler] düğmesine yapılan tıklama, ancak [Générer] düğmesine tıklanmasıyla başlatılan işlemin tamamlanmasından sonra işlenir. Az önce gördüğümüz gibi, bu işlem yürütme süresi boyunca olay döngüsünün iş parçacığını meşgul ederek [Annuler] düğmesine yapılan tıklamanın işlenmesini engelliyordu. Bu, Rx'in belirgin bir iyileştirme sağlayabileceği tipik durumlardan biridir;

2.6. Asenkron arayüz ve uygulaması

Şimdi [2] katmanının arayüzüne ve Rx ile gerçekleştirilmesine odaklanacağız. Bu, ilk bakışta anlaşılabilir olmayabilir. Amacımız, bu uygulamanın kodunun basitliğini vurgulamak.

Asenkron arayüz şu şekildedir:

public interface IRxService {
  // [a,b]'teki rastgele sayılar
  // n aralığında n rastgele sayı üretilir [minCount, maxCount]
  // sayılar, milisaniye gecikme süresinden sonra üretilir,
  // burada [delay], [minDelay, maxDelay] aralığında bir rastgele sayıdır
  public Observable<UiResponse> getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay, UiResponse uiResponse);
}

2.3. paragrafta sunulan senkron arayüzle arasındaki farklar şunlardır:

• 2.3. paragrafta sunulan [UiResponse] sınıfı artık [getAleas] yönteminin parametreleri arasında yer almaktadır (6. satır). Bunun nedeni, isteklerin artık paralel olarak yürütülmesi ve hizmetin sonucunu döndürmeden önce rastgele bir süre beklemesi nedeniyle, yanıtların bize istek sırasına göre gelmeyecek olmasıdır. Bu nedenle, diğer bilgilerin yanı sıra istek numarasını da içeren [UiResponse] nesnesini aktarıyoruz:

  // müşteri kimliği (istek)
  private int idClient;
  // hizmet yanıtı
  private ServiceResponse serviceResponse;
  // gözlem iş parçacığının adı
  private String observedOn;
  // istek saati
  private String requestAt;
  // yanıt saati
  private String responseAt;

• Asenkron hizmetin yanıt türü [Observable<UiResponse>] türüdür. [Observable<>] türü, Rx kütüphanesi tarafından sağlanır. [Observable<UiResponse>] türündeki sonuç, [getAleas] yönteminin [UiResponse] türünde bir değer akışı sağladığını ve bu değerlerin gözlemcisine tek tek aktarıldığını gösterir;

Şimdi bu arayüzün uygulamasına bir göz atalım:

public class RxService implements IRxService {

  // hizmet
  private IService service;

  // üretici
  public RxService(IService service) {
    this.service = service;
  }

  @Override
  public Observable<UiResponse> getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay, UiResponse uiResponse) {
    return Observable.create(subscriber -> {
      try {
        uiResponse.setServiceResponse(service.getAleas(a, b, minCount, maxCount, minDelay, maxDelay));
        subscriber.onNext(uiResponse);
      } catch (Exception e) {
        subscriber.onError(e);
      } finally {
        subscriber.onCompleted();
      }
    });
  }
}

• 7-9. satırlar: Oluşturucuya senkron [IService] arayüzüne bir referans verilir. Rastgele sayıların üretilmesi işini bu arayüz üstlenecektir;
• [getAleas] yöntemi tarafından döndürülen gözlemlenebilir, [Observable.create] statik yöntemi ile oluşturulur. Senkron bir uygulamadan asenkron bir uygulama oluşturmayı sağlayan bu yöntemdir;
• 13. satır: [Observable.create] statik yönteminin parametresi, burada bir [Subscriber] türünü parametre olarak alan bir lambda işlevidir; bu da yine bir Rx türüdür. Bir [Subscriber], gözlemlenebilirlerin akışına, yani asenkron olarak iletilen verilerin akışına abone olan bir nesnedir. Burada bu abonenin üç yöntemi kullanılmaktadır:
  • [Subscriber.onNext], ona bir veri iletmek için (16. satır);
  • [Subscriber.onError], aboneye bir istisna iletmek için (18. satır);
  • [Subscriber.onCompleted], aboneye veri akışının sona erdiğini bildirmek için (satır 20);

Aynı gözlemlenebilir nesneye birden fazla abone olabilir. Burada, 15-16. satırlarda üretilen tek bir veriye abone olan tek bir abone olacaktır. Veri, hizmetin senkron uygulaması tarafından üretilir (15. satır) ve aboneye iletilir (16. satır).

Tüm bunlar muhtemelen hâlâ belirsiz olsa da, hizmetin bu asenkron uygulamasının olağanüstü özlü yapısı dikkat çekicidir.

2.7. Asenkron çağrı

Şimdi, [swing] katmanının [2] hizmetine yaptığı senkron çağrı [5]'i ayrıntılı olarak inceleyeceğiz:

private void doGenerateWithRxService() {
        // bekleme başlangıcı
        beginWaiting();
        // rastgele sayılar isteniyor
        Observable<UiResponse> observables = Observable.empty();
        for (int i = 0; i < nbRequests; i++) {
            UiResponse uiResponse = new UiResponse();
            uiResponse.setIdClient(i);
            // zamanlayıcı
            int schedulerIndex = jComboBoxSchedulers.getSelectedIndex();
            switch (schedulerIndex) {
            case 0:
                observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.io()));
                break;
...
            }
        }
...
    }

• 6-10. satırlar: kullanıcı tarafından talep edilen [nbRequests] isteklerinin yürütülmesi;
• 7-8. satırlar: Asenkron hizmetin [getAleas] yönteminin (13. satır) ihtiyaç duyduğu [UiResponse] nesnesinin hazırlanması. Bu işlem esas olarak isteğin [idClient] numarasının kaydedilmesinden ibarettir;
• 13. satır: Asenkron hizmetin [getAleas] yöntemi çağrılır. Bu yöntem, bir [Observable<UiResponse>] nesnesi döndürür. Bu çağrı, senkron hizmeti henüz çağırmaz. Asenkron [getAleas] koduna geri dönelim:

  @Override
  public Observable<UiResponse> getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay, UiResponse uiResponse) {
    return Observable.create(subscriber -> {
      try {
        uiResponse.setServiceResponse(service.getAleas(a, b, minCount, maxCount, minDelay, maxDelay));
        subscriber.onNext(uiResponse);
      } catch (Exception e) {
        subscriber.onError(e);
      } finally {
        subscriber.onCompleted();
      }
    });
}

Senkron hizmeti çağıracak olan 4-11. satırlardaki kod, ancak bir abone kaydolduğunda yürütülür. Abone olmadığı sürece bu kod yürütülmez.

[doGenerateWithRxService] yöntemindeki koda geri dönelim:

• 5. satır: boş bir gözlemlenebilir nesne oluşturulur (hiçbir şey gözlemlenmez);
• 13. satır: Akışı, [nbRequests] istekleriyle ilişkili [nbRequests] asenkron akışlarının birleştirilmesi olacak bir gözlemlenebilir oluşturulur. Bu, iki asenkron akışı birleştirmeye olanak tanıyan [Observable.mergeWith] yöntemi ile elde edilir. Rx terminolojisinde, [mergeWith] bir akış operatörü olarak adlandırılır. Bu operatörlerin özelliği, işlemin sonucunun çoğu zaman yine bir [Observable] olmasıdır. Sonuç olarak, 17. satırdan sonra, [observables] değişkeni, asenkron hizmet tarafından verilen [nbRequests] asenkron yanıtlarından oluşan tek bir akışı ifade eder;
• 13. satır: birleştirme işlemi şu şekilde yazılabilirdi:

observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse));

ancak biz şöyle yazdık:

observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.io()));

Burada, [subscribeOn] operatörünü [rxService.getAleas] gözlemlenebilir üzerinde kullandık. Sıkça olduğu gibi, sonuç yine bir gözlemlenebilir. [subscribeOn] operatörü, gözlemlenebilirin bir [Scheduler] tarafından sağlanan bir iş parçacığında yürütülmesi gerektiğini belirtmeye olanak tanır. Farklı durumlara uygun çeşitli [Scheduler] seçenekleri mevcuttur. Grafik arayüzde, bunların etkilerini görmek için birkaçını önerdik:

Bu, aşağıdaki kodu verir:

    private void doGenerateWithRxService() {
        // bekleme başlangıcı
        beginWaiting();
        // rastgele sayılar isteniyor
        Observable<UiResponse> observables = Observable.empty();
        for (int i = 0; i < nbRequests; i++) {
            UiResponse uiResponse = new UiResponse();
            uiResponse.setIdClient(i);
            // zamanlayıcı
            int schedulerIndex = jComboBoxSchedulers.getSelectedIndex();
            switch (schedulerIndex) {
            case 0:
                observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.io()));
                break;
            case 1:
                observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.computation()));
                break;
            case 2:
                observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.newThread()));
                break;
            case 3:
                observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.trampoline()));
                break;
            case 4:
                observables = observables.mergeWith(rxService.getAleas(a, b, minCount, maxCount, minDelay, maxDelay, uiResponse).subscribeOn(Schedulers.immediate()));
                break;
            }
        }
...
}

12-14. satırlardaki koda geri dönelim. [Schedulers.io()] zamanlayıcısı, her bir gözlemlenebilir nesneye yeni bir iş parçacığı atar. Kodu takip edersek:

• 5. satır: boş bir gözlemlenebilir var;
• 13. satır, 1. yineleme: observables, [observable0/thread0] listesidir (observable0 gözlemlenebiliri, thread0 iş parçacığı üzerinde yürütülmüştür);
• 13. satır, 2. yineleme: observables, [observable0/thread0, observable1/thread1] listesidir;
• vb...

Sonuç olarak, 28. satırdan sonra, farklı [nbRequests] iş parçacıklarında çalışan [nbRequests] gözlemlenebilirlerinin birleştirilmesinden elde edilen bir gözlemlenebilir elde ederiz. Testlerde göreceğimiz gibi, tüm zamanlayıcılar bu şekilde çalışmaz.

Asenkron hizmet çağrı kodunu incelemeye devam edelim:

private void doGenerateWithRxService() {
        // bekleme başlangıcı
        beginWaiting();
        // rastgele sayılar isteniyor
        Observable<UiResponse> observables = Observable.empty();
        for (int i = 0; i < nbRequests; i++) {
        ...
        }
        // gözlemci
        observables = observables.observeOn(SwingScheduler.getInstance());
        // bu gözlemlenebilirler çalıştırılıyor
        subscriptions.add(observables.subscribe(uiResponse -> {
            updateUi(uiResponse);
        } , th -> {
            System.out.println(th);
            doCancel();
        } , this::doCancel));
    }

• 10. satıra geldiğimizde, tek bir gözlemlenebilir nesne olduğunu gördük; bu, kullanıcının seçtiği zamanlayıcıya bağlı olarak farklı [nbRequests] iş parçacıklarında çalışabilen veya çalışamayan [nbRequests] gözlemlenebilir nesnelerinin birleşimidir;
• 10. satır: [observeOn] operatörü, gözlemlenebilir nesneden gelen verilerin hangi iş parçacığında alınacağını belirtmeye olanak tanır; burada [nbRequests] türündeki nesneler söz konusudur. Bir Swing arayüzünde başka seçenek yoktur. Arayüzün tüm güncellemeleri, olay döngüsünün iş parçacığında yapılmalıdır. Burada, gözlemlenebilir nesnenin verileri bir Swing bileşeni olan JList içinde görüntülenecektir. [SwingScheduler.getInstance()] iş parçacığı, olay döngüsünün iş parçacığını temsil eder. [SwingScheduler] sınıfı, RxJava kütüphanesinden değil, RxSwing türev kütüphanesinden gelir;
• 12. satıra gelindiğinde, 10. satırdaki gözlemlenebilirin henüz bir abonesi olmadığı için senkron hizmet hala çağrılmamıştır. 12-17. satırlar, [subscribe] operatörü sayesinde ona bir abone atar. Bu operatörün parametreleri burada üç lambda fonksiyonudur:
  • ilki olan [uiResponse -> {updateUi(uiResponse);}], gözlemlenebilir tarafından üretilen [UiResponse] nesnelerinden birini parametre olarak kabul eder. Burada bu türden [nbRequests] nesnelerine sahip olacağımızı hatırlayalım. İlgili yöntem, burada updateUi, bu sonucu kullanmalıdır;
  • İkinci [th -> {System.out.println(th);doCancel();}], parametre olarak bir [Throwable] türünü kabul eder; bu durumda, gözlemlenebilirin yürütülmesi sırasında meydana gelen bir istisna söz konusudur. İlgili yöntem bu bilgiyi kullanmalıdır. Burada, bu bilgi konsola yazdırılır (15. satır) ve yürütme iptal edilir; bu da grafik arayüzün bazı öğelerinin güncellenmesine yol açar;
  • Üçüncü [this::doCancel], gözlemlenebilirin iletilecek verisi kalmadığını bildirdiğinde çağrılır. Burada gözlemlenebilir, [nbRequests] gözlemlenebilirlerinin birleşimidir. Sonuç gözlemlenebilir, onu oluşturan tüm gözlemlenebilirler kendi işlerini tamamladıklarını bildirdiklerinde, işinin bittiğini gösterecektir. Dolayısıyla bu üçüncü lambda fonksiyonu yürütüldüğünde, tüm veriler alınmış demektir. Yerel [doCancel] yöntemi, yürütmenin tamamlandığını yansıtmak üzere grafik arayüzü günceller;

[subscriptions] değişkeni şu şekilde tanımlanmıştır:

    // gözlemlenebilir değişkenlere abonelikler
protected List<Subscription> subscriptions = new ArrayList<Subscription>();

[Subscription] türü bir aboneliği temsil eder, yani bir abone ([Subscriber]) ile abonenin izlediği nesne ([Observable]) arasındaki bağlantıyı ifade eder. Burada bir abonelik listesi kullandık, ancak bu örnekte sadece bir tane var. Gözlemlenebilirin iletilecek verisi kalmadığını bildirdiğinde yürütülen yerel yöntem [doCancel] şu şekildedir:

    @Override
    protected void doCancel() {
        // bekleme sonu
        endWaiting();
        // abonelikler söz konusu olduğunda
        if (jCheckBoxRxSwing.isSelected() && subscriptions != null) {
            subscriptions.forEach(Subscription::unsubscribe);
        }
}

• 7. satır, gözlemlenebilirin tüm abonelerini iptal eder;

Bu özet açıklamadan aşağıdaki kilit noktalar çıkarılabilir:

• [Observable] türü, abonelere veya gözlemcilere tek tek gönderilen değerlerden oluşan bir akışı ifade eder;
• [Subscriber] türü, [Observable] türündeki bir aboneyi ifade eder;
• [Subscription] türü, bir aboneliği, yani bir [Subscriber] ile bir [Observable] arasındaki bağı ifade eder;
• [Observable] türü, çoğu gözlemlenebilirler üreten [mergeWith, empty, subscribeOn, observeOn, ...] operatörlerini kabul eder. Bu operatörler, gözlemlenebilirin yürütülmesinden önce onu yapılandırmak için kullanılır:
  • ne gözlemlenmek istendiği;
  • gözlemlenebilirin üzerinde çalışacağı iş parçacığı;
  • abonenin gözlemlenebilir nesneden verileri aldığı iş parçacığı;
• İki tür gözlemlenebilir vardır: [froid / cold] ve [chaud / hot]. Soğuk bir gözlemlenebilir, her yeni abone geldiğinde tamamen yürütülür. Her yürütme aynı verileri üretirse, her yeni abone bir öncekiyle aynı verileri alır. Sıcak bir gözlemlenebilir genellikle sürekli olarak veri üretir. Bir abone abone olduğunda, abonelik saatinden itibaren yayınlanan verileri alır. Daha önce yayınlanmış olabilecek verileri almaz. Örneğimizde gözlemlenebilir soğuktur: her yeni abone geldiğinde tamamen yeniden yürütülür. Örneğimizde gerçekte ne yürütülüyor? Bunu öğrenmek için gözlemlenen gözlemlenebilirin tanımına geri dönmeliyiz:

  @Override
  public Observable<UiResponse> getAleas(int a, int b, int minCount, int maxCount, int minDelay, int maxDelay, UiResponse uiResponse) {
    return Observable.create(subscriber -> {
      try {
        uiResponse.setServiceResponse(service.getAleas(a, b, minCount, maxCount, minDelay, maxDelay));
        subscriber.onNext(uiResponse);
      } catch (Exception e) {
        subscriber.onError(e);
      } finally {
        subscriber.onCompleted();
      }
    });
}

Her yeni abone için, [Observable.create] yönteminin parametresi olan lambda fonksiyonu (3. satır) yeniden çalıştırılır. Dolayısıyla, her yeni [subscriber] abonesi için 4-11. satırlar çalıştırılır;

2.8. Asenkron çağrıların test edilmesi

Öncelikle, önerilen farklı zamanlayıcıların etkisini göstereceğiz. Bunun için aşağıdaki parametreleri kullanıyoruz:

[1-2]'e küçük değerler atıyoruz; böylece istekler aynı iş parçacığında yürütülse bile çok uzun süre beklemek zorunda kalmayız.

2.8.1. [Schedulers.io] zamanlayıcısıyla

Aşağıdaki hususlara dikkat edilebilir:

• yanıtlar, isteklerin sırasına göre değil, farklı bir sırayla alınmaktadır (bkz. idClient);
• her istek farklı bir iş parçacığında gerçekleştirilmiştir;
• grafik arayüz bu sefer artık sabit değil:
  • sekmeler arasında geçiş yapılabiliyor;
  • verilerin geldiği görülebilir;
  • yürütme çok hızlı olduğu için [Annuler] düğmesini görmeye vaktimiz olmuyor. Bunu başka bir testte daha ayrıntılı olarak göstereceğiz;

2.8.2. [Schedulers.computation] zamanlayıcısıyla

Aşağıdaki noktalara dikkat çekilebilir:

• yanıtlar, isteklerin sırasına göre değil, farklı bir sırayla geliyor (bkz. idClient);
• istekler 8 iş parçacığında yürütülmüştür;
• 3 numaralı iş parçacığı, 8 ve 0 numaralı istekler için kullanılmıştır;
• 4 numaralı iş parçacığı, 9 ve 1 numaralı istekler için kullanılmıştır;
• diğer isteklerin her biri farklı bir iş parçacığına atanmıştır;

[Schedulers.computation] zamanlayıcısı, kullanılan makinede bulunan çekirdek sayısı kadar iş parçacığı kullanır. Bu bilgi, [Runtime.getRuntime().availableProcessors()] ifadesi ile elde edilir.

2.8.3. [Schedulers.newThread] zamanlayıcısıyla

[Schedulers.io] zamanlayıcısına benzer bir çalışma şekli söz konusudur.

2.8.4. [Schedulers.trampoline, Schedulers.immediate] zamanlayıcılarıyla

Çalışma senkron bir yapıdadır. Tüm istekler olay döngüsü iş parçacığı üzerinde yürütülür. Bu sonucu genelleştirmemek gerekir; sadece bu özel örnekte iki zamanlayıcının da senkron bir şekilde çalıştığını belirtmek yeterlidir.

2.9. Sınır durumlar

Bu örnekte, asenkron çalışmaya izin veren zamanlayıcılarla çalışacağız. Öncelikle, burada 8 iş parçacığıyla çalışan [Schedulers.computation] zamanlayıcısıyla istek sayısını 100'e çıkarıyoruz. Şu sonucu elde ediyoruz:

• [1]'te, [Annuler] düğmesi mevcut ve kullanılabilir durumdadır (asenkron çalışma);

Şimdi, yürütmeyi sonuna kadar devam ettirelim:

[2]'te, 100 sorgunun yürütülmesinin yaklaşık 4 saniye sürdüğü görülüyor (8 iş parçacığı üzerinde).

Şimdi, her isteği ayrı bir iş parçacığında çalıştıran [Schedulers.newThread] zamanlayıcısıyla aynı 100 isteği gerçekleştirelim:

[1]'te, 100 isteğin (100 iş parçacığı üzerinde) yürütülmesinin yarım saniye sürdüğü görülüyor. Dolayısıyla bu, [Schedulers.computation] zamanlayıcısına kıyasla belirgin şekilde daha hızlıdır.

Şimdi, yine aynı koşullarda ve yine [Schedulers.newThread] zamanlayıcısını kullanarak 800 istek yapalım. Şu sonuçları elde ediyoruz:

800 istek yaklaşık 1 saniye içinde gerçekleştiriliyor.

Bu sayıyı artırdığımızda (benim makinemde 2500'ün üzerinde istek - 1,5 saniyede gerçekleştirildi - bu sayı elbette yürütme anındaki çalışma ortamına büyük ölçüde bağlıdır), sonunda şu istisna ile karşılaşırız:

Yani bir yığın taşması söz konusu. Testler, [Schedulers.newThread] zamanlayıcısının çalışmasının deterministik olmadığını gösteriyor. Yukarıdaki istisna ortaya çıkabilir, yeni denemeler yapılabilir, ardından istisnayaya neden olan yapılandırmaya geri dönüldüğünde istisna bir daha ortaya çıkmayabilir.

2.10. Conclusion

Rx kütüphanesinin kullanımına dair bir örnek gösterdik. Öğrendiklerimizi özetleyelim:

Şu mimariyle başladık:

• [4]'te, [swing] katmanı, [service] katmanına senkron çağrılar yapıyordu;
• [5] katmanında, [swing] katmanı, [rxService] katmanına asenkron çağrılar yapıyordu; bu katman da [6] katmanını senkron bir şekilde [service] katmanına çağırıyordu;

İlk olarak gördüğümüz şey, Rx kütüphanesinin senkron [service] arayüzünden asenkron [rxService] arayüzünü kolayca oluşturmaya olanak sağladığıydı (bkz. paragraf 2.4). Bu önemli bir ders çünkü bu, senkron bir uygulamayı asenkron bir uygulamaya kolayca dönüştürebileceğimiz anlamına geliyor.

[swing] katmanında iki ayrı yöntem yazılmıştır:

• biri, servise senkron çağrılar yapmak için (bkz. paragraf 2.4);
• diğeri ise hizmete asenkron çağrılar yapmak için (bkz. paragraf 2.7);

Asenkron çağrıların yazımı, senkron çağrılarınkinden açıkça daha karmaşık çıktı. Bununla birlikte, senkronize edilmesi gereken birden fazla iş parçacığıyla eşzamanlı programlama yapmış olanlar, Rx çözümünün yazımının daha basit olduğunu ve zorlu sorunlar olan iş parçacıkları arasındaki senkronizasyon ve iletişim sorunlarının tümünü ortadan kaldırdığını göreceklerdir. Bu yazım sırasında, aşağıdaki önemli noktalara dikkat çektik:

• [Observable] türü, asenkron olabilen (ancak zorunlu olmayan) ve gözlemlenebilen olay (değer) akışını ifade eder;
• [Subscriber] türü, bir [Observable] türüne ait bir aboneyi ifade eder;
• [Subscription] türü, bir aboneliği, yani bir [Subscriber] ile bir [Observable] arasındaki bağı ifade eder;
• [Observable] türü, çoğu gözlemlenebilirler üreten [mergeWith, empty, subscribeOn, observeOn, ...] operatörlerini kabul eder. Bu operatörler, gözlemlenebilirin yürütülmesinden önce onu yapılandırmak için kullanılır:
  • ne gözlemlenmek istendiği;
  • gözlemlenebilirin üzerinde çalışacağı iş parçacığı;
  • abonenin gözlemlenebilir nesneden verileri aldığı iş parçacığı;
• İki tür gözlemlenebilir vardır: [froid / cold] ve [chaud / hot]. Soğuk bir gözlemlenebilir, her yeni abone geldiğinde tamamen yürütülür. Her yürütme aynı verileri üretirse, her yeni abone bir öncekiyle aynı verileri alır. Sıcak bir gözlemlenebilir genellikle sürekli olarak veri üretir. Bir abone abone olduğunda, abonelik saatinden itibaren yayınlanan verileri alır. Daha önce yayınlanmış olabilecek verileri almaz. Örneğimizde gözlemlenebilir soğuktur: her yeni abone geldiğinde tamamen yeniden yürütülür.

Rx kütüphanesinin faydalarını gösteren bir örnek inceledikten sonra, şimdi bu kütüphaneyi daha ayrıntılı olarak tanıtacağız.

Rx kütüphanesi, imza bölümlerinde jenerik parametreler içeren çok sayıda yönteme sahiptir. Bu imzalar hakkında kısa bir özet yapacağız (3. paragraf). Bu yöntemlerin parametreleri çoğunlukla işlevsel arayüzlerdir (Java 8), yani tek bir yönteme sahip arayüzlerdir. Dolayısıyla, gerçek parametreler bu arayüzlerin örnekleri olmalıdır. Java 8'den önce, bir arayüzü isimsiz bir sınıfla uygulamak yaygın bir uygulamaydı. Java 8 ile birlikte, arayüz fonksiyonel bir arayüz ise, onu bir lambda fonksiyonuyla uygulamak daha özlü bir yöntemdir. Bu nedenle bunları ele alacağız (4. paragraf). Bunu yaptıktan sonra, Java koleksiyonlarını lambda fonksiyonlarıyla işlemek için kullanılan [Stream] sınıfını (5. paragraf) tanıtacağız. Bu sınıf ilginçtir, çünkü RxJava sınıfındaki [Observable] sınıfı,

• bazı yöntemleri;
• aynı gözlemlenebilir nesneyi işlemek için yöntemleri birbirine zincirleme şekli;

Ardından, RxJava kütüphanesine özgü fonksiyonel arayüzleri tanıtacağız (6. paragraf). Daha sonra, Rx [Observable, Subscriber, Subscription, opérateurs] kütüphanesinin temel unsurlarına geçeceğiz (7. paragraf). [Observable] sınıfı, kendileri de birkaç kez aşırı yüklenmiş olan onlarca operatöre sahiptir. Bu durum başlangıçta büyük bir karmaşıklığa yol açar; çünkü bu operatörler ve aşırı yüklemeleri bazen sadece bir ayrıntıda farklılık gösterir ve deneyim olmadan hangi operatörün kullanılacağını bilmek zordur. Sadece sınırlı sayıda operatörü ele alacağız ve çoğu zaman bunların aşırı yüklemelerini göz ardı edeceğiz.

Önceki bölümün tamamı, basit konsol uygulamalarında RxJava kütüphanesi kullanılarak gerçekleştirilecektir. RxJava kütüphanesini öğrendikten sonra, bu kütüphaneyi iki tür grafik uygulamada kullanacağız:

• 8. paragrafta, örnek Swing uygulamasına geri dönerek daha ayrıntılı bir şekilde ele alacağız. Bu sırada RxSwing kütüphanesini kullanacağız;
• 9. paragrafta, RxAndroid kütüphanesini kullanarak bir Android uygulaması oluşturacağız;

Tüm bunlar tamamlandığında, okuyucu kendi başına yoluna devam edebilecek donanıma sahip olacaktır. Rx kütüphanesini sezgisel bir şekilde kullanabilmesi muhtemelen biraz zaman alacaktır. Bu kütüphaneyi özellikle ilgi çekici buldum. Ancak, anlaması zor geldi ve öğrenme süreci uzun sürdü. Umarım bu belge, okuyucu için bu süreci kısaltır. Bana göre bu çabaya değecek.