462. (RxAndroid) [Observable] [Schedulers] 스케줄러 사용해 subscribeOn , observeOn 스레드 관리 제어 실시

---

[개발 환경 설정]

​

개발 툴 : AndroidStudio

​

 

---

[설 명]

​

        /**
         * // -----------------------------------------
         * [요약 설명]
         * // -----------------------------------------
         * 1. 개발 환경 : 안드로이드 스튜디오
         * // -----------------------------------------
         * 2. Rx 관련 라이브러리 설치 (build.gradle) :
         *
         * implementation 'io.reactivex.rxjava3:rxandroid:3.0.0'
         * implementation 'io.reactivex.rxjava3:rxkotlin:3.0.1'
         * implementation 'io.reactivex.rxjava3:rxjava:3.0.7'
         *
         * // -----------------------------------------
         * 3. Rx 관련 설명 :
         *
         * ReactiveX 는 관찰가능한 절차를 통해 비동기, 이벤트 기반 프로그램을 구성하기 위한 라이브러리 입니다
         * ReactiveX 는 Observer Pattern 옵저버 패턴 을 확장하며, Sequence 를 조합할 수 있는 연산자를 지원 합니다
         * ReactiveX 는 low-level Thread, 동기화, Thread 안전성, non-blocking I/O에 관한 우려를 줄입니다
         * // -----------------------------------------
         * 4. Observable 구분 종류 설명 (Cold / Subject) :
         *
         * RxJava 의 Observable 에는 [Cold (차가운) : Observable] / [Hot (뜨거운) : Subject] 두 종류의 Obsevable 이 있습니다
         *
         * Observable / Subject 차이점 :
         *
         *   - [구독자 등록 기준] : [Observable] : 적어도 한 명의 관찰자가 있을 때 코드가 실행 (구독하기 전까지 데이터를 방출하지 않음)
         *
         *   - [구독자 수 기준] : [Observable] : 유니캐스트 (1:1 전송) 방식으로 옵저버 하나만을 구독 (subscribe) 할 수 있음
         *
         *   - [코드 실행 기준] : [Observable] : Observable 은 단지 하나의 함수이기 때문에 어떤 상태도 가지지 않으므로 모든 새로운 옵저버에 대해 관찰 가능한 create 코드를 반복해서 실행 (주요 버그와 비효율이 발생)
         *
         *   - [구독자 등록 기준] : [Subject] : 관찰자가 없더라도 코드가 실행되고 데이터가 발행 됨 (구독자의 존재 여부 관계없이 데이터 발행)
         *
         *   - [구독자 수 기준] : [Subject] : 멀티캐스트 (1:N 전송) 방식으로 여러개 옵저버를 구독 (subscribe) 할 수 있음
         *
         *   - [코드 실행 기준] : [Subject] : Subject 는 관찰자 세부 정보를 저장하고 코드를 한 번만 실행해 모든 관찰자에게 결과를 제공함
         * // -----------------------------------------
         * 5. Scheduler (스케줄러) 기본 개념
         *
         * Scheduler 는 반응형 프로그래밍에서 동시성을 관리하며, 스레드 관리를 제어하기 위한 subscribeOn , observeOn 메소드 가 있습니다
         * Scheduler 관련 용어 설명 :
         *   - AndroidSchedulers.mainThread() : 안드로이드의 UI 스레드에서 동작
         *   - Schedulers.io() : 비동기 처리를 위해 사용하는 스케줄러 (API 호출 등 네트워크를 사용한 호출)
         *   - Schedulers.computation() : 이벤트 그룹에서 간단한 연산이나 콜백 처리를 위해 사용 (스레드 풀이 순환하면서 실행됨)
         *   - Schedulers.from(executor) : 특정 executor를 스케줄러로 사용
         *   - Schedulers.newThread() : 새로운 스레드를 만드는 스케줄러
         *   - Schedulers.trampoline() : 큐에 있는 일이 끝나면 이어서 현재 스레드에서 수행
         *   - HandlerScheduler.from(handler) : 특정 핸들러 handler 에 의존하여 동작
         * // -----------------------------------------
         * */

 

---

[소스 코드]

​

        // [로직 처리 실시]
        try {

            // [Observable create 사용해 생성 실시 : 생성자]
            Observable<String> observable = Observable.create(subscriber -> {
                try {
                    if (subscriber.isDisposed() == false){ // [연결된 구독자가 있는 경우]
                        Log.i("---","---");
                        Log.w("//===========//","================================================");
                        Log.i("","\n"+"[A_Test >> main() :: create() : Send Data]");
                        Log.w("//===========//","================================================");
                        Log.i("---","---");

                        // [onNext] 데이터 알림 전달
                        subscriber.onNext("TWOK");

                        // [onCompleted] 완료 알림 전달
                        subscriber.onComplete();
                    }
                }
                catch (Exception e){
                    e.printStackTrace();

                    // [onError] 에러 알림 전달
                    subscriber.onError(e);
                }
            });


            // [Schedulers 설정 및 구독 결과 확인]
            observable
                    .subscribeOn(AndroidSchedulers.mainThread()) // [Observable (생성자) 로직을 IO 스레드에서 실행 : 백그라운드]
                    .observeOn(Schedulers.io()) // [Observer (관찰자) 로직을 메인 스레드에서 실행]
                    .subscribe(new Observer<String>() { // [Observable.create 타입 지정]
                        @Override
                        public void onSubscribe(@NonNull Disposable d) {
                            Log.i("---","---");
                            Log.d("//===========//","================================================");
                            Log.i("","\n"+"[A_Test >> main() :: onSubscribe() : 구독 해제 상태 확인]");
                            Log.i("","\n"+"[구독 해제 여부 : "+String.valueOf(d.isDisposed())+"]");
                            Log.d("//===========//","================================================");
                            Log.i("---","---");
                        }

                        @Override
                        public void onNext(@NonNull String s) {
                            Log.i("---","---");
                            Log.w("//===========//","================================================");
                            Log.i("","\n"+"[A_Test >> main() :: onNext() : 데이터 확인]");
                            Log.i("","\n"+"[value : "+String.valueOf(s)+"]");
                            Log.w("//===========//","================================================");
                            Log.i("---","---");
                        }

                        @Override
                        public void onError(@NonNull Throwable e) {
                            Log.i("---","---");
                            Log.e("//===========//","================================================");
                            Log.i("","\n"+"[A_Test >> main() :: onError() : 에러 발생]");
                            Log.i("","\n"+"[error : "+String.valueOf(e.getMessage())+"]");
                            Log.e("//===========//","================================================");
                            Log.i("---","---");
                        }

                        @Override
                        public void onComplete() {
                            Log.i("---","---");
                            Log.w("//===========//","================================================");
                            Log.i("","\n"+"[A_Test >> main() :: onComplete() : 완료 상태]");
                            Log.w("//===========//","================================================");
                            Log.i("---","---");
                        }
                    });
        }
        catch (Exception e){
            e.printStackTrace();
        }

 

---

[결과 출력]

 

D///===========//: ================================================
I/: [A_Test >> main() :: onSubscribe() : 구독 해제 상태 확인]
I/: [구독 해제 여부 : false]
D///===========//: ================================================


W///===========//: ================================================
I/: [A_Test >> main() :: create() : Send Data]
W///===========//: ================================================


W///===========//: ================================================
I/: [A_Test >> main() :: onNext() : 데이터 확인]
I/: [value : TWOK]
W///===========//: ================================================


W///===========//: ================================================
I/: [A_Test >> main() :: onComplete() : 완료 상태]
W///===========//: ================================================

 

---

​