RxJS’e yakından bakış -3: Scheduler

By Tahir Kardak On Nov 7, 2019

Merhaba bu yazımda RxJS’in bir diğer bileşenlerinden biri olan Scheduler’den bahsedeceğim. Bu yazı ile, bir sonraki yazımız olan RxJS operatörlerine geçmeden önce RxJS’e çekirdeğinde bulunan tüm konseptlerden bahsetmiş olacağım.

Önceki Bölümler

https://bilisim.io/2019/10/08/rxjse-merhaba-deyin/

https://bilisim.io/2019/10/20/rxjse-yakindan-bakis-1-observables/

https://bilisim.io/2019/11/02/rxjse-yakindan-bakis-2-subjects/

Scheduler Nedir?

Bir aboneliğin(subscription) ne zaman başlayacağını ve bildirimlerin ne zaman yapılacağını kontrol eden yapıdır. Başlamadan önce faydalarından bahsedeyim:

Umulmadık buglardan kurtarır
Yazdığınız kodun yanlış sırada çalışmasını engeller

Makalenin sonunda gerçek hayattan bir örnek vermeye çalışacağım.

Scheduler, üç bileşenden oluşur.

Execution Context: Görevin nerede ve ne zaman çalıştırılacağını (yürütüleceğini) gösterir.

Sync Tasks
Macro Tasks -> setTimeout()
Micro Tasks -> Promise()
AnimationFrame -> animationFrame

Execution Policy: Öncelik veya diğer ölçütlere göre görevlerin nasıl saklanacağı ve hangi sırayla alınacağından sorumludur.

Clock: Scheduler’da bulunan now() getter metodu ile zaman kavramını verir. Eğer bir scheduler, belirli bir zamanda bir işlem yapmak durumundaysa zaman kavramına ihtiyacı vardır.

Scheduler, bir observable’ın kendisini dinleyen observer’a hangi yürütme bağlamında(execution context) bildirim göndereceğini tanımlamanıza izin verir.

Aşağıdaki örnekte; 1, 2, 3 değerlerini eş zamanlı olarak yayan bir observable mevcut. Bu değerleri iletmek için kullanılacak eşzamansız programlayıcıyı belirlemek için observeOn operatörü kullanılıyor. https://stackblitz.com/edit/typescript-pzv3t6



import { Observable, asyncScheduler } from 'rxjs';
import { observeOn } from 'rxjs/operators';


const observable = new Observable((observer) => {
  observer.next(1);
  observer.next(2);
  observer.next(3);
  observer.complete();
});
console.log('***********************');
console.log('Scheduler Kullanılmayan Observable');
console.log('abonelik başlamadan Önce');
observable.subscribe({
  next(x) {
    console.log('Gelen değer: ' + x)
  },
  error(err) {
    console.error('Hata oluştu: ' + err);
  },
  complete() {
     console.log('bitti');
  }
});
console.log('abonelik bittikten sonra');
console.log('***********************');


const observable1 = new Observable((observer) => {
  observer.next(1);
  observer.next(2);
  observer.next(3);
  observer.complete();
}).pipe(
  observeOn(asyncScheduler)
);

console.log('Scheduler Kullanılan Observable');
console.log('Abonelik başlamadan önce');
observable1.subscribe({
  next(x) {
    console.log('Gelen değer: ' + x)
  },
  error(err) {
    console.error('Hata oluştu: ' + err);
  },
  complete() {
     console.log('bitti');
  }
});
console.log('Abonelik bittikten sonra');

import { Observable, asyncScheduler } from 'rxjs';

import { observeOn } from 'rxjs/operators';

const observable = new Observable((observer) => {

observer.next(1);

observer.next(2);

observer.next(3);

observer.complete();

});

console.log('***********************');

console.log('Scheduler Kullanılmayan Observable');

console.log('abonelik başlamadan Önce');

observable.subscribe({

next(x) {

console.log('Gelen değer: ' + x)

error(err) {

console.error('Hata oluştu: ' + err);

complete() {

console.log('bitti');

}

});

console.log('abonelik bittikten sonra');

console.log('***********************');

const observable1 = new Observable((observer) => {

observer.next(1);

observer.next(2);

observer.next(3);

observer.complete();

}).pipe(

observeOn(asyncScheduler)

);

console.log('Scheduler Kullanılan Observable');

console.log('Abonelik başlamadan önce');

observable1.subscribe({

next(x) {

console.log('Gelen değer: ' + x)

error(err) {

console.error('Hata oluştu: ' + err);

complete() {

console.log('bitti');

}

});

console.log('Abonelik bittikten sonra');

Örneğimizde iki tane observable mevcut. İlk observable’a abone olurken observeOn operatörü kullanılmıyor. Konsola yazılan çıktıda herşey eş zamanlı bir şekilde yazılıyor. İkinci observable’da ise observeOn kullanıyor konsol çıktısında eş zamansız olarak değerler görüntüleniyor. Bunun sebebi observeOn(asyncScheduler);new Observable ile Son Observer arasında proxy Observer tanıtmasıdır. Aşağıdaki örnek kodda, bu durum daha anlaşılır bir şekilde gösterilmiştir. https://stackblitz.com/edit/fvbm14



import { Observable, asyncScheduler } from 'rxjs';
import { observeOn } from 'rxjs/operators';

var observable = new Observable((proxyObserver) => {
  proxyObserver.next(1);
  proxyObserver.next(2);
  proxyObserver.next(3);
  proxyObserver.complete();
}).pipe(
  observeOn(asyncScheduler)
);

var finalObserver = {
  next(x) {
    console.log('Gelen Değer' + x)
  },
  error(err) {
    console.error('Hata: ' + err);
  },
  complete() {
     console.log('bitti');
  }
};

console.log('Abonelikten önce');
observable.subscribe(finalObserver);
console.log('abonelikten sonra');

import { Observable, asyncScheduler } from 'rxjs';

import { observeOn } from 'rxjs/operators';

var observable = new Observable((proxyObserver) => {

proxyObserver.next(1);

proxyObserver.next(2);

proxyObserver.next(3);

proxyObserver.complete();

}).pipe(

observeOn(asyncScheduler)

);

var finalObserver = {

next(x) {

console.log('Gelen Değer' + x)

error(err) {

console.error('Hata: ' + err);

complete() {

console.log('bitti');

}

};

console.log('Abonelikten önce');

observable.subscribe(finalObserver);

console.log('abonelikten sonra');

observeOn() operatörü gecikme için bir parametre daha alır. Bu parametre ile yayılma işlemi herhangi bir zamana planlanabilir(schedule). Eğer observeOn()operatörüne herhangi bir değer verilmezse varsayılan olarak 0 şeklinde programlanır. Yukarıdaki örneklerde observeOn()operatörüne gecikme parametresi verilmediği için varsayılan olarak 0 olarak alınmıştır. Aşağıdaki örnekte ise abonelik başladıktan üç saniye sonra değerlerin yayılmasını ayarlayan scheduler kullanımı mevuttur. https://stackblitz.com/edit/rxjs-scheduler-kullanimi



import { asyncScheduler, range } from 'rxjs';
import { observeOn } from 'rxjs/operators';

range(0,10)
.pipe(observeOn(asyncScheduler,3000))
.subscribe(console.log);

import { asyncScheduler, range } from 'rxjs';

import { observeOn } from 'rxjs/operators';

range(0,10)

.pipe(observeOn(asyncScheduler,3000))

.subscribe(console.log);

Scheduler Türleri

asyncScheduler; RxJs tarafından sunulan built-in scheduler’dir. Aşağıdaki schedulerlerin her biri, Scheduler nesnesinin statik özellikleri kullanılarak oluşturulabilir .



+-------------------------+-------------------------------------------------------------+
|        SCHEDULER        |                       KULLANIM AMACI                        |
+-------------------------+-------------------------------------------------------------+
| null                    | Herhangi bir zamanlayıcı geçilmezse;                        |
|                         | bildirimler eş zamanlı ve tekrarlı olarak yapılır.          |
|                         | Bunu sabit zamanlı işlemler veya                            |
|                         | yinelimeli kuyruk işlemleri için kullanın                   |
+-------------------------+-------------------------------------------------------------+
| queueScheduler          | Bir kuyruktaki o anda  olay çerçevesinini planlar.          |
|                         | Tekrarlama işlemlerinde bunu kullanın.                      |
+-------------------------+-------------------------------------------------------------+
| asapScheduler           | Micro taslar için kullanılır.                               |
|                         | Temelde mevcut işten sonra, ancak bir sonraki işten önce.   |
|                         | Eşzamansız dönüşümler için bunu kullanın.                   |
+-------------------------+-------------------------------------------------------------+
| asyncScheduler          | setInterval ile birlikte kullanılır.                        |
|                         | Zaman tabanlı işlemlerde kullanın.                          |
+-------------------------+-------------------------------------------------------------+
| animationFrameScheduler | Bir sonraki browser içeriği yeniden boyamadan               |
|                         | hemen önce gerçekleşecek olan görevlerde kullanılır.        |
|                         | Akıcı browser animasyonları oluşturmak için kullanılabilir. |
+-------------------------+-------------------------------------------------------------+

+-------------------------+-------------------------------------------------------------+

| SCHEDULER | KULLANIM AMACI |

+-------------------------+-------------------------------------------------------------+

| null | Herhangi bir zamanlayıcı geçilmezse; |

| | bildirimler eş zamanlı ve tekrarlı olarak yapılır. |

| | Bunu sabit zamanlı işlemler veya |

| | yinelimeli kuyruk işlemleri için kullanın |

+-------------------------+-------------------------------------------------------------+

| queueScheduler | Bir kuyruktaki o anda olay çerçevesinini planlar. |

| | Tekrarlama işlemlerinde bunu kullanın. |

+-------------------------+-------------------------------------------------------------+

| asapScheduler | Micro taslar için kullanılır. |

| | Temelde mevcut işten sonra, ancak bir sonraki işten önce. |

| | Eşzamansız dönüşümler için bunu kullanın. |

+-------------------------+-------------------------------------------------------------+

| asyncScheduler | setInterval ile birlikte kullanılır. |

| | Zaman tabanlı işlemlerde kullanın. |

+-------------------------+-------------------------------------------------------------+

| animationFrameScheduler | Bir sonraki browser içeriği yeniden boyamadan |

| | hemen önce gerçekleşecek olan görevlerde kullanılır. |

| | Akıcı browser animasyonları oluşturmak için kullanılabilir. |

+-------------------------+-------------------------------------------------------------+



import { asyncScheduler, of, animationFrameScheduler ,asapScheduler,queueScheduler} from 'rxjs';
import { observeOn } from 'rxjs/operators';

const gecikme =1;
const obs$ = of(1);
obs$.pipe(observeOn(animationFrameScheduler,gecikme))
.subscribe((x)=> console.log('5. animationFrameScheduler:'+x));

obs$.pipe(observeOn(asyncScheduler,gecikme))
.subscribe((x)=> console.log('4. asyncScheduler:'+x));

obs$.pipe(observeOn(asapScheduler,gecikme))
.subscribe((x)=> console.log('3. asapScheduler:'+x));

obs$.pipe(observeOn(queueScheduler,gecikme))
.subscribe((x)=> console.log('1. queueScheduler:'+x));

obs$.subscribe((x)=> console.log('2. Normal:'+x));

import { asyncScheduler, of, animationFrameScheduler ,asapScheduler,queueScheduler} from 'rxjs';

import { observeOn } from 'rxjs/operators';

const gecikme =1;

const obs$ = of(1);

obs$.pipe(observeOn(animationFrameScheduler,gecikme))

.subscribe((x)=> console.log('5. animationFrameScheduler:'+x));

obs$.pipe(observeOn(asyncScheduler,gecikme))

.subscribe((x)=> console.log('4. asyncScheduler:'+x));

obs$.pipe(observeOn(asapScheduler,gecikme))

.subscribe((x)=> console.log('3. asapScheduler:'+x));

obs$.pipe(observeOn(queueScheduler,gecikme))

.subscribe((x)=> console.log('1. queueScheduler:'+x));

obs$.subscribe((x)=> console.log('2. Normal:'+x));

https://stackblitz.com/edit/rxjs-scheduler-kullanimi-neb9zz

Yukarıdaki örnekte gecikme değişkenini 0 yaptığınızda farklı çıktı almaktasınız.

Peki bu bilgi gerçek hayatta ne işime yarayacak?

Konuyu daha iyi anlamak için bir örnek yapalım. Aşağıdaki Angular uygulamasındaki hataya bakalım. Uygulama çalıştığında ExpressionChangedAfterItHasBeenCheckedError hatası alınmaktadır. Hatanın çözümü için üç farklı yöntem gösterilecektir. Üçüncü yöntem RxJS Scheduler ile yapılmıştır. https://stackblitz.com/edit/angular-owpbyp



import { Component , Input, AfterViewInit } from '@angular/core';

@Component({
  selector: 'my-app',
  template: `<h4>Merhaba {{name}}!</h4>`,
  styles: [`h1 { font-family: Lato; }`]
})
export class AppComponent implements AfterViewInit  {
   @Input() name: string;

  ngAfterViewInit() {
    this.name = 'Dünya';
  }
}

import { Component , Input, AfterViewInit } from '@angular/core';

@Component({

selector: 'my-app',

styles: [`h1 { font-family: Lato; }`]

})

export class AppComponent implements AfterViewInit {

@Input() name: string;

ngAfterViewInit() {

this.name = 'Dünya';

}

Bu hata ne zaman ortaya çıkar

Eğer kodunuzda AfterViewInit yaşam döngüsü kancasını(lifecycle hook) kullanıyorsanız, ViewChild, AfterViewInit çağrılana kadar tanımsızdır.
DOM’u doğrudan değiştiriyorsanız, Angular değişiklikleri tespit edemez ve doğru reaksiyon veremez.
HTML şablonunuzun içerisinde fonksiyon çağırıyorsanız race condition durumunun ortaya çıkmasında meydana gelir

Bu hatadan kurtulmak için;

ngAfterViewInit içersinde setTimeout kullanılabilir. https://stackblitz.com/edit/angular-zjxnbh



import { Component , Input, AfterViewInit } from '@angular/core';

@Component({
  selector: 'my-app',
  template: `<h4>Merhaba {{name}}!</h4>`,
  styles: [`h1 { font-family: Lato; }`]
})
export class AppComponent implements AfterViewInit  {
   @Input() name: string;

  ngAfterViewInit() {
setTimeout(() => {
      this.name = 'foo';
    });
  }
}

import { Component , Input, AfterViewInit } from '@angular/core';

@Component({

selector: 'my-app',

styles: [`h1 { font-family: Lato; }`]

})

export class AppComponent implements AfterViewInit {

@Input() name: string;

ngAfterViewInit() {

setTimeout(() => {

this.name = 'foo';

});

}

2. Change Detection kullanılarak bu sorun çözülebilir. https://stackblitz.com/edit/angular-rhtgfb



import { Component , Input, AfterViewInit ,ChangeDetectorRef} from '@angular/core';

@Component({
  selector: 'my-app',
  template: `<h4>Merhaba {{name}}!</h4>`,
  styles: [`h1 { font-family: Lato; }`]
})
export class AppComponent implements AfterViewInit  {
   @Input() name: string;

  constructor(private cdr: ChangeDetectorRef) { }


  ngAfterViewInit() {
    this.name = 'foo';
    this.cdr.detectChanges();
  }
}

import { Component , Input, AfterViewInit ,ChangeDetectorRef} from '@angular/core';

@Component({

selector: 'my-app',

styles: [`h1 { font-family: Lato; }`]

})

export class AppComponent implements AfterViewInit {

@Input() name: string;

constructor(private cdr: ChangeDetectorRef) { }

ngAfterViewInit() {

this.name = 'foo';

this.cdr.detectChanges();

}

3. RxjS ile daha şık bir işlem ile çözülebilir. https://stackblitz.com/edit/angular-mjqtms



import { Component , Input, AfterViewInit} from '@angular/core';
import { asyncScheduler } from 'rxjs';

@Component({
  selector: 'my-app',
  template: `<h4>Merhaba {{name}}!</h4>`,
  styles: [`h1 { font-family: Lato; }`]
})
export class AppComponent implements AfterViewInit  {
   @Input() name: string;

  ngAfterViewInit() {
     asyncScheduler.schedule(() => {
      this.name = 'foo';
    });
    
  }
}

import { Component , Input, AfterViewInit} from '@angular/core';

import { asyncScheduler } from 'rxjs';

@Component({

selector: 'my-app',

styles: [`h1 { font-family: Lato; }`]

})

export class AppComponent implements AfterViewInit {

@Input() name: string;

ngAfterViewInit() {

asyncScheduler.schedule(() => {

this.name = 'foo';

});

}

`asyncScheduler kullanırsanız; scheduler metoduna verilen callback fonksiyonu ()=> { this.name = 'foo';}, setTimeout gibi varsayılan değerde eşzamansız olarak yürütülür. Sonuc olarak; bir adet makro görev zamanlanmış oldu.

Konuyla ilgili olarak aşağıdaki videoyu tavsiye ederim

Kaynak

RxJS