RxJava 와 같은 Reactive Extensions 잘 쓰시나요? (Exception 처리 질문)

Question

기존에 Couchbase 관련 작업을 하다가 맛뵈기로만 살짝 사용을 하다가 최근에 몇 가지 일처리 때문에 RxJava 을 소스 코드 몇 군데에서 쓰고 있습니다. 그런데, 딸리는 영어와 실력 때문에 사용하면서 힘든 점이 있네요. 비동기 처리이긴 한데...모든 처리가 끝난 뒤 다음 행으로 가는 것인지, 아니면 callback 와 같이 처리가 되는 건지부터 막히기 시작해서, 최근에는 Exception 처리 때문에 막 헷갈려하고 있습니다. 특히 subscribe() 에서 &#64;Transactional 이 연결된 Service 의 method 을 호출해서 처리하는게 있는데, RuntimeException 을 상속받은 Exception 을 만들어서 처리하고 있습니다. 그런데 이걸 다시 Observable 외부로 보내줘야 이놈을 다시 rollback 처리를 할텐데...애매모호 하더라구요. 저의 경우는 그냥 try { Observable.from(array).subscribe(s -> { // Code }); } catch (Exception e) { Throwable throwable &#61; e.getCause(); if (throwable instanceof CustomException) { throw (CustomException) throwable; } else { throw new Exception(throwable); } } 이런 식으로 처리를 해주고 있습니다. 그런데 이게 맞는 걸까요? 아무리 메뉴얼이나 다른걸 찾아봐도 Exception 처리에 대한 부분에 대한 예제도 없고 설명도 별로 없어서...되게 찝찝하게 처리하고 있습니다. 혹시 RxJava 쓰고 계신 분 계신가요?

손이시렵다 · Answer

Rx자바는 잘 모르지만 폰트는 엄청 깔끔하네요윈도우에서도 사용가능한가요?

zepinos · Answer

손이시렵다 님 // 그럼요. 지금 올린 것도 윈도우즈에서 캡처한 것입니다. 네이버 나눔고딕코딩입니다. ^^;;

손이시렵다 · Answer

아 이게 그거군요 감사합니다

star16m · Answer

본업에서 사용한적은 없지만..좀 더 스마트한 방법이 있을 듯 합니다.subscribe 안에 s -> 이런 기술은 1.7 혹은 1.8 에서 가능한거죠??아직 낡은 jdk 만 만지고 있어서 따라가지가 않네요.ㅎㅎ아무튼 s -> 이건 onNext 이고e -> 요건 onError 일 겁니다.이렇게 하지 마시고 아래와 같이 해보세요.new Subscriber<String>() { &#64;Override public void onCompleted() { log(&#34;completed&#34;); } &#64;Override public void onError(Throwable e) { // error log(&#34;error:&#34; &#43; e); } &#64;Override public void onNext(String s) { log(&#34;on next: &#34; &#43; s); }}

zepinos · Answer

star16m 님 // 조언 고맙습니다. 말씀하신대로 s -> 이 부분은 람다이고, Java 8 부터 지원합니다. 두번째 매개변수로 말씀하신 대로 onError 을 받을 수 있는데, 저기서 로그를 찍는게 목적이 아니라(로그를 찍는 거라면 여기서 처리하는게 맞습니다) 다시 상위로 throw 을 해줘야 합니다. 그래야 Observable 이 속해있는 method 의 &#64;Transactional 이 제대로 동작합니다. 그런데, onError() 에서 throw 을 하면 다시 Observable 의 Throwable 로 래핑되어서(OnErrorFailedException) 밖으로 나갑니다. 그래서 그냥 바로 외부에서 try/catch 을 해버린 것입니다. 저기서 throw 을 했을 때 원하는 행태대로 throw 가 안됐던 것으로 기억합니다. RuntimeException 은 throw 가 되는데, Exception 은 throw 가 안됩니다. 저도 람다에 대한 이해가 빈약하다 보니...개념이 좀 헷갈리네요.

star16m · Answer

음.. onError 에 이미 CustomException 이 Wrapping 되어 들어 오는데 이걸 instanceof 해서 다시 throw 하면 어떨까요??

zepinos · Answer

onError 안에서 getCause() 로 해서 CustomException 을 cast 해서 throw 하면 다시 OnErrorFailedException 으로 쌓여서 외부로 나갑니다.

star16m · Answer

자꾸 이것저것 해보시라는 것밖엔 안되서 좀 죄송하긴 하지만아래 사이트에서 call 과 함께 각 sequence 메소드를 제어하는 방법으로 가능할 것 같습니다.http://rakuishi.com/archives/rxjava/일본어라 보기 힘드시겠지만 요부분입니다. Observable .create(new Observable.OnSubscribe() { @Override public void call(Subscriber subscriber) { Request request = new Request.Builder() .url("http://rakuishi.com/index.xml") .get() .build(); OkHttpClient okHttpClient = new OkHttpClient(); try { Response response = okHttpClient.newCall(request).execute(); subscriber.onNext(response); subscriber.onCompleted(); } catch (IOException e) { subscriber.onError(e); } } })

star16m · Answer

저기서 subscriber.onError(e) 와 함께 그대로 throw 하면 그것까지 캐치는 안할 것 같아요

zepinos · Answer

네 조언 고맙습니다. 저 방식대로 하면...병렬로 될지 모르겠네요. 다른 분들이 보실지 모르겠지만, 팁 아닌 팁 드리면, Observable 을 그냥 사용하면 for 랑 동일하게 직렬로 동작합니다. 다음과 같은 형태대로 구현해야 병렬로 돌아가요. (물론 이게 좋은 형태인지는 모르겠지만, 어쨌든 병렬로는 돌아갑니다) 병렬로 도는게 다가 아니라, Thread 에서 join() 을 쓴 것처럼 병렬로 된 것들이 완료되면 다음 라인을 처리하는 형태가 되는거죠. 이 구조에서 저런 형태의 에러 처리가 될지 한 번 고민해보겠습니다. 조언해주셔서 너무 고맙습니다. Observable .just(&#34;1&#34;, &#34;2&#34;, &#34;3&#34;) .flatMap(s -> Observable.defer(() -> { // 코드 return Observable.just(s); }).subscribeOn(Schedulers.computation())) .toList().toBlocking().single();

RxJava 와 같은 Reactive Extensions 잘 쓰시나요? (Exception 처리 질문)

10개의 답변

Q&A 목록