RxJava를 이용한 Reactive Programming

What is RxJava?

생명주기와 동시성 프로그래밍

기존의 AsyncTask는 많은 성능저하를 가지고 온다. 또한 생명주기에 따라 인스턴스를 해제해주지 않으면 memory leak에 대한 주의가 필요합니다. 이러한 문제점을 해결하고자한 프로그래밍 패러다임이 Reactive 프로그래밍입니다. 예로 Reactive 프로그래밍을 간단하게 말하면 각 항목을 관찰(Observe)하다가 항목에 변화가 있으면 등록된 Observer에 따라 갱신해주는 구성입니다.

Reactive 프로그래밍의 대표적인 예는 엑셀차트입니다.

Observable -> 관측 가능한 값의 흐름을 나타내는 객체 (생산자) 버튼 클릭 subscribe 함수를 통해 구독이 가능하며 , 구독자에게 값의 흐름을 알려준다.

Observer - > 값의 흐름을 받는 객체 (소비자) 텍스트뷰 Observable 값의 흐름에 따라 onNext, onError, onCompelete를 호출 받는다.

기존 비동기처리의 코드는 아래의 형태로 구현됩니다.

MyThread extends Thread{
  interface MyCallback{
    void onFinishDownloadImage(T t);
    void onError(Exception e);
  }

  MyCallback callback;

  @Override
  public void run(){
    try {
      //download image
    } catch {
        callback.onError(e);
    }
      runOnUiThread(new Runnable(){
        @override
        public void run(){
          callback.onFinishDownloadImage(bitmap);
        }
      });
  }}

동시성을 처리하는 방법으론 안드로이드의 기본적인 동시성을 다루는 Handler는 기본적으로 uiThread를 관리합니다. 특징으로는 Looper 함께 사용되며 Looper는 메세지 큐를 가지고 있기 때문에 호출되는 순서로 실행됩니다. syncronize같은 저수준 키워드를 이용해서 critical section을 관리할 필요가 없습니다. 기존의 비동기 처리 방식의 AsyncTask의 단점을 뽑아보자면

@Overrideprotected void onPreExecute(){
  progressBar.setVisibility(View.VISIBLE);}

과

@Overrideprotected void onPostExecute{Bitmap bitmap}{
  imageView.setImageBitmap(bitmap);
  progressBar.setVisibility(View.GONE);}

는 메인스레드에서 호출되지만

@Overrideproteted Bitmap doInBackground(String... params) {
  try {
      bitmap = downloadBitmap(params);
  } catch (Exception e) { ... }
  return bitmap;}

doInBackground는 백그라운드에서 호출되는데 만약 예외가 발생했다면 ui에 통지할 방법이 없습니다. 만약 핸들러나 액티비티를 참조하게 되면 메모리의 누수가 발생합니다.

복잡한 요구사항을 지닌 로직을 예로 들어봅시다.

• 어떤 데이터가 있는데 -> Observable
• 어떤 쓰레드 에서 수행하고 -> subscribeOn(background)
• 필요한 데이터만 걸러내고 -> filter(paidUser())
• 데이터를 어떻게 변환하고 -> map(StringToJson())
• 오류 발생시에는 어떻게 대응할지 -> doOnError(handleError())
• 어떤 쓰레드에서 결과를 받을지, -> observeOn(mainTread)
• 시작하면, -> subscribe()->subscription
• 필요한 경우 취소 할수 있어야 한다. subscription.unsubscribe()
• 테스트도 쉬워야겠고! assert(paiduser()), assert(stringToJson())

이와 같은 가독성과 테스트의 용이성으로 인해 로직에만 집중 할수 있습니다. 추가로 가독성을 위해 람다표현식의 도입을 고려할 수도 있습니다.

Runnable task = new Runnable(){
   @Override
   public void run(){
      system.out.println("Hello Lamda");
   }};

위와 같은 코드를 람다표현식을 이용해서 아래와 같이 변경할 수 있습니다.

Runnable task2 = () -> {System.out.println("Hello Lamda");};

안드로이드에서 표현식을 이용하기 위해선 Retrom lambda 플러그인을 추가하면 사용할 수 있습니다.

why? 왜 어째서 사용 하는가?

Plain old java concurrent programming

동시성 프로그래밍을 위한 기존의 스레드에는 불편함이 있습니다. 아래코드는 기존의 코드 중 비트맵을 다운 받는 예입니다.

new Thread{
  @Override
    public void run(){
      /To-do
      URL url = new URL(strUrl);
      HttpURLConnection urlConnetction = ...(생략);
      urlConnetction.connet();
      iStream = urlConnection.getInputStream();
      bitmap = BitmapFactory.decodeStream(iStream);

      /* 비트맵을 보여줘야 하기 때문에 ui조정이 필요하다 따라서
      핸들러를 사용하거나 runOnUiThread를 사용 해야 한다.*/
      runOnUiThread(new Runnable()){
          @Override public void run(){
            imageView.setImageBitmap(bitmap);
          }
      }
    }}

하지만 다운로드 받는 도중에 에러가 발생할 수 있다. 그러므로 try-catch사용하여 예상지 못한 예외를 처리해야 합니다. 추가로 다운로드가 실패했을때 다운로드가 실패한것을 보여주기위한 Toast를 출력해보도록 합니다.

new Thread{
  @Override
    public void run(){

      try {
      URL url = new URL(strUrl);
      HttpURLConnection urlConnetction = ...(생략);
      urlConnetction.connet();
      iStream = urlConnection.getInputStream();
      bitmap = BitmapFactory.decodeStream(iStream);
      } catch(Exception e) {
          log(e.toString);

          runOnUiThread(new Runnable(){
               showToast("다운로드 실패");
              });
      }
      runOnUiThread(new Runnable(){
          @Override public void run(){
            imageView.setImageBitmap(bitmap);
          });
      }
    }}

네트워크를 통한 처리는 저장소가 부족할수도 있고 인터넷이 안될수도 있고 인터넷이 연결되는데 어떠한 문제로 인해 실패할수도 있습니다 이로 인해 try catch문이 복잡해지고 가독성이 현저히 떨어지게 됩니다. 리팩토링을 했지만 이 메소드가 어떤 스레드에서 도는지 명시되있지 않습니다.

Bitmap downloadUrl(String strUrl){
    try {
        iStream = getInputStream(strUrl);
    } catch (Exceoption e) {
        handleExceoption(e, ()-> {"다운로드 실패"});
    } fianlly {
        close(iStream);
    }
    return BitmapFactory.decodeStream(iStream);}

private void handleExeption(Exception e,OnError onError){
          ..생략}

Bitmap downloadUrl(String strUrl){
    ...생략}

//1번void handleExeption(Exception e,OnError onError){
    Lod.d(onError.call());}

//2번 에러를 메인 스레드에서 처리해서 다소 유연해보이지 않을수 있다.void handleExeption(Exception e,OnError onError){
    runOnUiThread(new Runnable(){
        void run(){
          Lod.d(onError.call());
        }
      }
    )}

하지만 뷰를 조작 하는 코드가 있을 경우에는 1번은 예외가 발생하고 2번은 정상 작동 합니다. 뷰를 그리는 도중 뷰의 상태가 바뀌기 때문에 calledFromWrongThreadException 예외가 발생합니다.

Bitmap downloadUrl(String strUrl){
        handleExceoption(e, ()-> {
          Toas.makeText(getActiity(), ()-> {
            textView.setText("다운로드 실패")
          })
        });}

//1번 -> calledFromWrongThreadExceptionvoid handleExeption(Exception e,OnError onError){
    Lod.d(onError.call());}

//2번void handleExeption(Exception e,OnError onError){
    runOnUiThread(new Runnable(){
        void run(){
          Lod.d(onError.call());
        }
      }
    )}

위와 같은 예외발생을 방지하기 위해 동작하는 스레드를 명시해주었습니다.

//1번 -> calledFromWrongThreadExceptionvoid handleExeptionOnUiThread(Exception e,OnError onError){
    Lod.d(onError.call());}

//2번void handleExeptionOnBgThread(Exception e,OnError onError){
    runOnUiThread(new Runnable(){
        void run(){
          Lod.d(onError.call());
        }
    }}

RxJava를 이용한 예를 하나 더 들어보겠습니다. final PublicSubject inputText = publishSubject.create(); TextView autoCompleteTextView = (TextView)findViewById(R.id.auto_complete_text);

1초 동안 사용자 입력이 없는 경우 네트워크에서 값을 받아와서 갱신 1초동안 입력이 없는 경우 작업을 시작 하게 되고 io thread에서 observeOn함수를 통해 처리하는데 flatamp 을 통해 인터넷에서 비동기로 다운을 받아 다받아지면 메인 스레드에서 구독을하고 subscibe 에서 view에 없데이트 해준다.

inputText.throttleWithTimeout(1,TimeUnit.SECONDS)
   .observerOn(Schdulers.io())
   .flatMap(new GetAutocompleteKwyswordsFromNetwork())
   .observeOn(AndroidSchdulers.mainThread())
   .subscribe(new SetTextAciton(autoCompleteTextView));
   

복잡한 비동기처리를 비교적 간단하게 처리할 수 있는 것을 볼 수 있습니다. RxJava뿐만 아니라 다양한 언어로 구현이 되있습니다.

Language	Library
Java	RxJava
JavaScript	RxJS
Scala	RxScala
Swift	RxSwift
C#	Rx.NET
C#(Unity)	UniRx
Clojure	RxClojure
C++	RxCpp
Ruby	Rx.rb
Python	RxPY
Groovy	RxGroovy
JRuby	RxJRuby
Kotlin	RxKotlin
PHP	RxPHP

위의 표와 같이 다양한 언어로 구현되어있는 라이브러리를 이용해 비동기처리의 복잡성을 푸는 데 도움이 될 수 있습니다.

참고 : https://deview.kr/2014/session?seq=4, LIVEO 기술 블로그