[STUDY] RxAndroid Part 7 (Scheduler 이해하기)

Android 개발/RxAndroid 2017. 5. 13. 15:42
336x280(권장), 300x250(권장), 250x250, 200x200 크기의 광고 코드만 넣을 수 있습니다.

 

[목차]==================================================

1. RxJava와 RxAndroid에서 제공하는 Scheduler 종류

1) RxJava가 제공하는 Scheduler

2) RxAndroid는 제공하는 Scheduler

2. 독립적으로 사용 가능한 Scheduler

3. 사용법 예제

1) Scheduler 안한 경우 기본 동작
2) subscribeOn() API만 사용한 경우 동작
3) observerOn() API만 사용한 경우 동작
4) subscribeOn() & observerOn() 1번씩 API 사용한 경우 동작
5) subscribeOn() & observerOn() 다수 API 사용한 경우 동작

======================================================

 

 

RxAndroid에서는 Scheduler를 통해 어느 쓰레드에서 실행이 될지 결정 할 수 있습니다.
Scheduler는 subsctibeOn(), observeOn() 에서 각각 지정할 수 있는데

subsctibeOn()은 observable의 작업을 시작하는 쓰레드를 선택 할 수 있습니다.( 중복해서 적을 경우 가장 마지막에 적힌 스레드에서 시작합니다.)

observeOn()은 이후에 나오는 Operator, subscribe의 Scheduler를 변경 할 수 있습니다.

 

1. RxJava와 RxAndroid에서 제공하는 Scheduler 종류

1) RxJava가 제공하는 Scheduler

  • Schedulers.computation()

이벤트 룹에서 간단한 연산이나 콜백 처리를 위해서 쓰는 것입니다. I/O 처리를 여기에서 해서는 안됩니다.

RxComputationThreadPool라는 별도의 스레드 풀에서 돌아갑니다. 최대 cpu갯수 ?개의 스레드 풀이 순환하면서 실행됩니다.

  • Schedulers.from(executor)

특정 executor를 스케쥴러로 사용합니다

  • Schedulers.immediate

현재 스레드에서 즉시 수행합니다.
observeOn()이 여러번 쓰였을 경우 immediate()를 선언한 바로 윗쪽의 스레드를 따라갑니다.

  • Schedulers.io()

동기 I/O를 별도로 처리시켜 비동기 효율을 얻기 위한 스케줄러입니다. 자체적인 스레드 풀에 의존합니다.
자체적인 스레드 풀 CachedThreadPool을 사용합니다. API 호출 등 네트워크를 사용한 호출 시 사용됩니다.

  • Schedulers.newThread()

항상 새로운 스레드를 만드는 스케쥴러입니다.

  • Schedulers.trampoline()

큐에 있는 일이 끝나면 이어서 현재 스레드에서 수행하는 스케쥴러

 

※ 일부 오퍼레이터들은 자체적으로 어떤 스케쥴러를 사용할지 지정합니다. 예를 들어 buffer 오퍼레이터는 Schedulers.computation()에 의존하며 repeat은 Schedulers.trampoline()를 사용합니다.

 

 

2) RxAndroid는 제공하는 Scheduler

  • AndroidSchedulers.mainThread()

안드로이드의 UI 스레드에서 동작

  • HandlerScheduler.from(handler)

특정 핸들러 handler에 의존하여 동작

 

※ 안드로이드에 특화된 스케쥴러입니다. 보통은 RxAndroid가 제공하는 AndroidSchedulers.mainThread()와 RxJava가 제공하는 Schedulers.io()를 조합해서 Schedulers.io()에서 수행한 결과를 AndroidSchedulers.mainThread()에서 받아 UI에 반영하는 패턴등이 일반적으로 쓰입니다.

 

 

2. 독립적으로 사용 가능한 Scheduler

Scheduler는 Observable, Operator, Subscriber 모델 밖에서 별도로 사용할 수 있습니다.

 worker = Schedulers.newThread().createWorker();


worker.schedule(new Action0() {

    @Override
    public void call() {
        realmJob();
    }

});

 

 Scheduler.Worker worker = Schedulers.newThread().createWorker();
        worker.schedule(new Runnable() {
            @Override
            public void run() {
                log("worker: " + Thread.currentThread().getName());
            }
        });

[출력결과]
worker: RxNewThreadScheduler-1

 


3. 사용법 예제

1) Scheduler 안한 경우 기본 동작

Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                log("subscribe:" + Thread.currentThread().getName());
                e.onNext("next");
            }
        }).map(s1 -> {
            log("map: " + Thread.currentThread().getName());
            return s1;
        }).subscribe(s -> log("on next: " + Thread.currentThread().getName()));

[출력결과]
subscribe:main
map: main
on next: main


2) subscribeOn() API만 사용한 경우 동작

Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                log("subscribe:" + Thread.currentThread().getName());
                e.onNext("next");
            }
        }).subscribeOn(Schedulers.computation()).map(s1 -> {
            log("map: " + Thread.currentThread().getName());
            return s1;
        }).subscribe(s -> log("on next: " + Thread.currentThread().getName()));

[출력결과]
subscribe:RxComputationThreadPool-1
map: RxComputationThreadPool-1
on next: RxComputationThreadPool-1


3) observerOn() API만 사용한 경우 동작

Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                log("subscribe:" + Thread.currentThread().getName());
                e.onNext("next");
            }
        }).observeOn(Schedulers.computation()).map(s1 -> {
            log("map: " + Thread.currentThread().getName());
            return s1;
        }).subscribe(s -> log("on next: " + Thread.currentThread().getName()));

[출력결과]
subscribe:main
map: RxComputationThreadPool-1
on next: RxComputationThreadPool-1


4) subscribeOn() & observerOn() 1번씩 API 사용한 경우 동작

Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                log("subscribe:" + Thread.currentThread().getName());
                e.onNext("next");
            }
        }).subscribeOn(Schedulers.io())
              .observeOn(AndroidSchedulers.mainThread())
                .map(s1 -> {
                        log("map: " + Thread.currentThread().getName());
                        return s1;
         }).subscribe(s -> log("on next: " + Thread.currentThread().getName()));

[출력결과]
subscribe:RxCachedThreadScheduler-1
map: main
on next: main


5) subscribeOn() & observerOn() 다수 API 사용한 경우 동작

 Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(ObservableEmitter<String> e) throws Exception {
                log("subscribe:" + Thread.currentThread().getName());
                e.onNext("next");
            }
        }).subscribeOn(Schedulers.io())
          .subscribeOn(Schedulers.computation())
          .observeOn(AndroidSchedulers.mainThread()).map(s1 -> {
                        log("map1: " + Thread.currentThread().getName());
                        return s1;
         }).observeOn(Schedulers.newThread()).map(s2 -> {
                        log("map2: " + Thread.currentThread().getName());
                        return s2;
         }).observeOn(Schedulers.single()).map(s3 -> {
                        log("map3: " + Thread.currentThread().getName());
                        return s3;
         }).subscribe(s -> log("on next: " + Thread.currentThread().getName()));

[출력결과]
subscribe:RxCachedThreadScheduler-1
map1: main
map2: RxNewThreadScheduler-1
map3: RxSingleScheduler-1
on next: RxSingleScheduler-1

 



 

출처 : 인터넷에서 RxAndroid 검색하여 필요한 정보를 다양한 사이트에서 종합하여 작성된 것입니다. 많은 사이트 내용을 종합하여 공부하여 작성하다보니 일일이 나열하지 못하였습니다. ㅈㅅ(_ _) 이글은 자유롭게 퍼 가셔서 도움이 되었으면 좋겠습니다. 감사합니다. 

 

저작자표시 (새창열림)

'Android 개발 > RxAndroid' 카테고리의 다른 글

[STUDY] RxAndroid Part 9 (Connectable Observable )  (0) 2017.05.13
[STUDY] RxAndroid Part 8 (Error 핸들링 방법)  (0) 2017.05.13
[STUDY] RxAndroid Part 6 (Subject 사용 - PublishSubject/BehaviorSubject)  (0) 2017.05.13
[STUDY] RxAndroid Part 5 (Observable 변형/가공/합성 - map/flatMap/zip)  (0) 2017.05.13
[STUDY] RxAndroid Part 4 (Observable - Create/Defer/Empty/Never/Throw/Never/Throw/From/Interval 등)  (0) 2017.05.13
블로그 이미지

차봉규

IT개발 공부하려고 만든 블로그 입니다^^ 부족한점 많더라고...도움 많이 부탁해요

,

[STUDY] RxAndroid Part 2 (개념 - Gradle 설정/Rx핵심개념 : Observable,Subscriber, Subscription , Scheduler )

Android 개발/RxAndroid 2017. 5. 13. 15:32
336x280(권장), 300x250(권장), 250x250, 200x200 크기의 광고 코드만 넣을 수 있습니다.

[목차]==================================================

1. Gradle 설정

2. Rx의 핵심개념

Observable

Subscriber

Subscription

Scheduler

======================================================

 


 

1. Gradle 설정

안드로이드 스튜디오에서 프로젝트를 만들고 app 디렉토리의 build.gradle 파일을 열어 봅시다.

dependencies {

    compile fileTree(dir: 'libs', include: ['*.jar'])

    testCompile 'junit:junit:4.12'

    compile 'com.android.support:appcompat-v7:23.1.1'

    compile 'io.reactivex:rxandroid:1.1.0'

}

Gradle 설정에서 가장 중요한 부분은 의존성입니다.

안드로이드도 자바 환경이기 때문에 rxjava를 포함하지 않는 것에 의아할 수 있습니다. RxAndorid는 RxJava에 대한 의존성을 가지고 있고 RxAndroid를 의존성에 포함하면 안드로이드 개발 환경에 문제가 없습니다. 혹시 새로운 버전을 적용하길 원한다면 명시적으로 지정할 수 있습니다.

지금은 Rxandroid는 1.1.0 이고 Rxjava는 1.1.1 이라 별도로 의존성을 추가해 줬다.
그리고 build.gradle 파일을 동기화 해주면 적용은 끝난다. 
1.0 이하 버전에서는 각종 컴포넌트의 이벤트를 처리 부분이 포함되어 있었으나 
라이브러리의 경량화와 유지보수등의 문제로 서드파티 라이브러리로 빼버렸다고 한다. 아래 외에도 다양한 라이브러리가 있으니 gitHub를 찾아보면 됨


RxLifecycle - Lifecycle handling APIs for Android apps using RxJava
RxBinding - RxJava binding APIs for Android's UI widgets.
SqlBrite - A lightweight wrapper around SQLiteOpenHelper and ContentResolver which introduces reactive stream semantics to queries.
Android-ReactiveLocation - Library that wraps location play services API boilerplate with a reactive friendly API.
rx-preferences - Reactive SharedPreferences for Android

2. Rx의 핵심개념

  • Observable : 아이템을 발행하는 일
  • Subscriber(Observer) : 발행한 아이템을 소비 / Subscriber는 onNext() , onComplete(), onError()를 각각 가지고 있는데, onNext()는 여러번 호출 될 수 있고 이후에 onComplete(), onError()이 처리 / Observable 안에는 Subscriber를 가지고 있어서 onNext를 호출하면 다음에 있는 Observable 이나 Subscriber가 받을 수 있음
  • Subscription : Observable이 방출한 아이템을 Observer가 구독해서 반응을 합니다. 바로 이 때 Observable과 Observer의 연결을 Subscription을 통해서 이루어 집니다.
  • Scheduler : 멀티쓰레드, 비동기 방식 등에 유용하며, Scheduler는 해당 Observable, Operator, Subscriber를 어떤 스레드에서 수행할지 결정하는 것입니다. 스케줄러가 어떤 부분을 맞게 되는지는 subscribeOn과 observeOn으로 지정합니다
  • Observable과 Subscriber를 주목

데이터의 강을 만드는 Observable과 강에서 데이터를 하나씩 건지는 Subscriber가 리액티브 프로그래밍의 가장 핵심적인 요소입니다. Observable은 데이터를 제공하는 생산자로 세가지 유형의 행동을 합니다.

onNext - 새로운 데이터를 전달한다.

onCompleted - 스트림의 종료.

onError - 에러 신호를 전달한다


package io.realm.simpleobservable;


import android.os.Bundle;

import android.support.v7.app.ActionBarActivity;

import android.util.Log;

import android.widget.TextView;


import rx.Observable;

import rx.Subscriber;


public class MainActivity extends ActionBarActivity {

    private static final String TAG = MainActivity.class.getName();


    @Override

    protected void onCreate(Bundle savedInstanceState) {


        super.onCreate(savedInstanceState);

        setContentView(R.layout.activity_main);


        Observable<String> simpleObservable =

                Observable.create(new Observable.OnSubscribe<String>() {

                    @Override

                    public void call(Subscriber<? super String> subscriber) {

                        subscriber.onNext("Hello RxAndroid !!");

                        subscriber.onCompleted();

                    }

                });


        simpleObservable

                .subscribe(new Subscriber<String>() {

                    @Overridea

                    public void onCompleted() {

                        Log.d(TAG, "complete!");

                    }


                    @Override

                    public void onError(Throwable e) {

                        Log.e(TAG, "error: " + e.getMessage());

                    }


                    @Override

                    public void onNext(String text) {

                        ((TextView) findViewById(R.id.textView)).setText(text);

                    }

                });

    }

}


 

 

 

출처 : 인터넷에서 RxAndroid 검색하여 필요한 정보를 다양한 사이트에서 종합하여 작성된 것입니다. 많은 사이트 내용을 종합하여 공부하여 작성하다보니 일일이 나열하지 못하였습니다. ㅈㅅ(_ _) 이글은 자유롭게 퍼 가셔서 도움이 되었으면 좋겠습니다. 감사합니다. 

저작자표시 (새창열림)

'Android 개발 > RxAndroid' 카테고리의 다른 글

[STUDY] RxAndroid Part 6 (Subject 사용 - PublishSubject/BehaviorSubject)  (0) 2017.05.13
[STUDY] RxAndroid Part 5 (Observable 변형/가공/합성 - map/flatMap/zip)  (0) 2017.05.13
[STUDY] RxAndroid Part 4 (Observable - Create/Defer/Empty/Never/Throw/Never/Throw/From/Interval 등)  (0) 2017.05.13
[STUDY] RxAndroid Part 3 (메모리 누수 - RxLifecycle, compose / CompositeSubscription)  (0) 2017.05.13
[STUDY] RxAndroid Part 1 (개요 - Rx등장/이해 등)  (0) 2017.05.13
블로그 이미지

차봉규

IT개발 공부하려고 만든 블로그 입니다^^ 부족한점 많더라고...도움 많이 부탁해요

,

공지사항

최근에 받은 트랙백

글 보관함

달력

«   2025/05   »
일 월 화 수 목 금 토
1 2 3
4 5 6 7 8 9 10
11 12 13 14 15 16 17
18 19 20 21 22 23 24
25 26 27 28 29 30 31

티스토리툴바