RxJava: Wrapping Listener Callback with Observable Examples

이 글에서는 콜백 기반 리스너 스타일 API를 RxJava 2를 사용하여 반응형 API로 변환하는 방법을 보여드리겠습니다.

콘텐츠

소개
목표
시작하기
소스 만들기
- 리스팅: BasicSource.java
- 리스팅: UnboundSource.java
옵저버블로 전환
- 목록: Observables.java
인스턴스화 및 구독
출력
해설: RxJava 배우기
포스트 스크립트

소개

반응형 프로그래밍과 RxJava는 특히 지난 몇 년 동안 꽤 화제가 되었습니다. 저는 얼마 전 안드로이드 프로젝트에서 RxJava를 사용해 보았습니다. 다른 방법으로도 충분히 쉽게 해결할 수 있는 간단한 스레딩 문제가 있었습니다. 하지만 RxJava에 대한 글을 읽은 적이 있었기 때문에 사용해 보기로 했습니다. 코드가 훨씬 더 간단하고 이해하기 쉬워서 즉시 깊은 인상을 받았습니다.

이러한 성공에도 불구하고 RxJava는 배우기 어렵다는 평판이 있습니다. 최근 프로젝트에서 저는 데이터베이스의 실시간 업데이트를 처리하고 싶었습니다. 데이터베이스(카우치베이스 라이트)에는 변경 사항을 모니터링하기 위한 콜백 기반 시스템이 있습니다. 저는 이 콜백을 리액티브 구조로 감싸고 싶었습니다. (이것은 관찰 가능 또는 유동적. 둘 중 하나를 선택하는 방법에 대한 후속 기사를 찾아보세요.)

제가 가장 먼저 발견한 것은 제가 원하는 일반적인 버전의 좋은 예를 찾을 수 없다는 것이었습니다. 간단한 예는 RxJava 문서와 비슷하지만 피하고 싶은 몇 가지 단점이 있습니다. 예를 들어, 이 예제에서는 Event 객체에 주어진 이벤트가 스트림의 마지막 이벤트인지 확인하는 메서드가 있다고 가정합니다. 안드로이드의 많은 콜백에는 이러한 메서드가 없습니다.

나중에 발견했지만 스택 오버플로 게시물 기본을 꽤 잘 다루고 있어서 더 자세히 알고 싶었습니다.

내부에 대한 전체 논의는 당연히 책 한 권을 채울 수 있습니다. 이 글에서는 핵심만 다루겠습니다. 자세한 내용을 이해하는 데 도움이 되는 몇 가지 실험이 포함된 코드가 있습니다. 한 번에 설명하기에는 너무 방대하므로 다음 기회에 다루도록 하겠습니다.

목표

좀 더 명확하게 설명하기 위해 이벤트 중심 프로그래밍에서 흔히 사용되는 리스너 콜백 인터페이스를 가져와서 관찰 가능.

즉, 다음에서 어떻게 이동합니까?

public interface OnItemListener<T> {
  void onItem(T item);
}

public 인터페이스 OnItemListener<T> {

void onItem(T 항목);

}

에

Observable<T>

관찰 가능<T>

Android의 온클릭 리스너 는 이러한 유형의 API의 한 예입니다. 온클릭 리스너 는 하나의 메서드가 있는 인터페이스입니다, onClick. Android 보기 를 매개변수로 설정합니다. 안드로이드 시스템은 이를 사용하여 버튼 누르기 등의 이벤트 스트림을 전달합니다.

시작하기

이 작업의 소스는 GitHub에서 찾을 수 있습니다. 여기.

이 글에서는 해당 코드의 일부만 살펴보겠습니다. 코드의 다른 부분은 다양한 실험을 시도하기 위해 설계되었습니다. 이러한 내용은 다음 글에서 다룰 수 있습니다. 여기서는 핵심 주제에만 집중하겠습니다.

따라 하려면 리포지토리를 복제하세요. 이 코드는 gradle로 빌드되도록 구성되어 있으므로 명령줄에서 실행하거나 원하는 환경으로 가져올 수 있습니다.

소스 만들기

이 객체는 일종의 이벤트 소스를 수신하기 위한 API를 변환하는 것입니다. 가장 먼저 필요한 것은 테스트할 실제 이벤트 스트림입니다. 실험을 위해 빌드된 몇 가지 소스가 있으므로 기본 클래스부터 시작하겠습니다.

리스팅: BasicSource.java

// File: src/main/java/com/couchbase/rx/BasicSource.java
package com.couchbase.rx;

public class BasicSource<T> {
  protected volatile OnItemListener<? super T> listener;

  public static interface OnItemListener<T> { void onItem(T item); }

  public void setOnItemListener(OnItemListener<? super T> listener) { this.listener = listener; }
}

// 파일: src/main/java/com/couchbase/rx/BasicSource.java

패키지 com.카우치베이스.rx;

public 클래스 기본 소스<T> {

보호됨 휘발성 OnItemListener<? super T> 리스너;

public 정적 인터페이스 OnItemListener<T> { void onItem(T 항목); }

public void 설정온아이템 리스너(OnItemListener<? super T> 리스너) { 이.리스너 = 리스너; }

}

이것은 인터페이스를 정의하며 모든 구현에 필요한 공통 필드 및 세터 메서드를 가지고 있습니다.

다음으로, 다음과 같이 무제한 이벤트 스트림을 모방하는 소스를 도출합니다.

리스팅: UnboundSource.java

// File: src/main/java/com/couchbase/rx/UnboundSource.java
package com.couchbase.rx;

import java.util.function.Supplier;

import static com.couchbase.Util.sleep;

public class UnboundSource<T> extends BasicSource<T> {
  private Supplier<T> supplier;
  private OnItemListener<? super T> current;

  public UnboundSource(Supplier<T> supplier) {
    this.supplier = supplier;
  }

  public void start() {
    System.out.println("Source emitting on thread " + Thread.currentThread().getName());

    for (;;) {
      System.out.println("Emitting item on thread " + Thread.currentThread().getName());

      current = listener;

      if (null != current) current.onItem(supplier.get());

      sleep(100);
    }
  }
}

// 파일: src/main/java/com/couchbase/rx/UnboundSource.java

패키지 com.카우치베이스.rx;

가져오기 자바.활용.함수.공급업체;

가져오기 정적 com.카우치베이스.Util.수면;

public 클래스 UnboundSource<T> 확장 기본 소스<T> {

비공개 공급업체<T> 공급업체;

비공개 OnItemListener<? super T> 현재;

public UnboundSource(공급업체<T> 공급업체) {

이.공급업체 = 공급업체;

}

public void 시작() {

시스템.out.println("스레드에서 소스 방출" + 스레드.currentThread().getName());

에 대한 (;;) {

시스템.out.println("스레드에서 항목 내보내기" + 스레드.currentThread().getName());

현재 = 리스너;

만약 (null != 현재) 현재.onItem(공급업체.get());

수면(100);

}

이 버전은 다음을 사용하여 새 항목을 생성합니다. 공급업체 함수를 생성자에 전달합니다. 이는 공급자가 무엇이든 생성할 수 있기 때문에 실제 출력되는 객체에는 특별한 것이 없음을 보여줄 뿐입니다.

명시적으로 항목 생성을 시작하는 방법이 있습니다. 여기서는 무한 루프를 사용하여 무한히 생성합니다.

할당 리스너 에 현재 는 구독을 폐기하는 것이 null 확인과 실제 호출 사이에 발생할 수 있는 경쟁 조건에서 작동합니다. onItem 콜백.

옵저버블로 전환

이제 개방형 일련의 버튼 클릭을 모방하는 소스가 생겼습니다. 다음으로 사용자 정의 관찰 가능.

RxJava 문서에서 권장하는 방법을 사용하겠습니다. 여기서는 Observable.create 메서드를 서브클래싱하는 대신 관찰 가능 를 직접 호출할 수 있습니다. (프로젝트에는 비교를 위해 후자를 수행하는 코드도 포함되어 있습니다.)

목록을 살펴보세요.

목록: Observables.java

// File: src/main/java/com/couchbase/rx/Observables.java
package com.couchbase.rx;

import io.reactivex.Observable;
import io.reactivex.ObservableOnSubscribe;

import com.couchbase.Util.ComputeFunction;

public class Observables {
  public static void main(String[] args) {
    UnboundSource<Object> source = new UnboundSource<>(Object::new);

    ObservableOnSubscribe<Object> handler = emitter -> {
      System.out.println("Create on thread - " + Thread.currentThread().getName());

      source.setOnItemListener(item -> {
        System.out.println("Listen on thread - " + Thread.currentThread().getName());

        if (emitter.isDisposed()) return;
        emitter.onNext(item);
      });

      emitter.setCancellable(() -> source.setOnItemListener(null));
      source.start();
    };

    Observable.create(handler)
      .subscribe(ComputeFunction::compute, Throwable::printStackTrace,
        () -> System.out.println("Done"), t -> System.out.println("onSubscribe on thread " + Thread.currentThread().getName()));
  }
}

// 파일: src/main/java/com/couchbase/rx/Observables.java

패키지 com.카우치베이스.rx;

가져오기 io.리액티브엑스.관찰 가능;

가져오기 io.리액티브엑스.옵저버블온구독;

가져오기 com.카우치베이스.Util.컴퓨트 함수;

public 클래스 관찰 가능 항목 {

public 정적 void 메인(문자열[] args) {

UnboundSource<Object> 출처 = new UnboundSource<>(개체::new);

옵저버블온구독<Object> 핸들러 = emitter -> {

시스템.out.println("스레드에서 만들기 - " + 스레드.currentThread().getName());

출처.설정온아이템 리스너(항목 -> {

시스템.out.println("스레드에서 듣기 - " + 스레드.currentThread().getName());

만약 (emitter.isDisposed()) 반환;

emitter.onNext(항목);

});

emitter.설정 취소 가능(() -> 출처.설정온아이템 리스너(null));

출처.시작();

};

관찰 가능.create(핸들러)

.구독(컴퓨트 함수::계산, 던지기 가능::프린트스택트레이스,

() -> 시스템.out.println("완료"), t -> 시스템.out.println("on스레드에서 구독" + 스레드.currentThread().getName()));

}

먼저, 우리의 UnboundSource.

다음으로, 다음과 같은 인스턴스를 생성합니다. 옵저버블온구독 람다 표현식을 사용합니다. 재정의하는 메서드입니다, 구독에는 하나의 매개변수인 이미터 객체입니다.

이는 두 번째 람다 표현식을 통해 소스의 리스너 콜백과 구독자 간의 연결을 제공합니다. 이 두 번째 표현식은 emitter.isDisposed 를 호출하여 구독이 여전히 활성 상태인지 확인한 다음, 항목을 다운스트림으로 푸시합니다. emitter.onNext. 이것이 바로 지금까지 쌓아온 핵심 라인입니다.

원래 콜백을 연결했으니 이제 흐름을 멈출 수 있는 방법을 제공하고자 합니다. 우리는 취소 가능 를 사용하겠습니다. 람다 표현식은 리스너 콜백을 무효화하여 흐름을 끊습니다.

A 일회용 도 작동합니다. 이 스택 오버플로 답변 를 참조하면 둘의 차이점과 어떤 것을 사용할지 선택하는 방법에 대해 잘 알 수 있습니다.

일회용품은 스트림 구독을 취소하는 RxJava 2의 솔루션입니다. 이 게시물 에서 일회용품 사용에 대한 일반적인 이유를 설명합니다.

모든 것이 상호 연결된 상태에서 소스를 실행하여 이벤트 생성을 시작합니다.

인스턴스화 및 구독

당사의 옵저버블온구독 인스턴스를 생성할 수 있습니다. 관찰 가능 인스턴스를 호출하여 create.

그리고 관찰 가능 클래스는 꽤 많은 메서드를 사용할 수 있는 유창한 인터페이스를 제공합니다. 'onXXX' 함수를 개별 조각으로 분리하는 메서드를 사용하여 결과 옵저버블을 구독합니다.

출력

그림과 같이 예제를 실행하면 다음과 같은 출력이 표시됩니다.

onSubscribe on thread main
Create on thread - main
Source emitting on thread main
Emitting item on thread main
Listen on thread - main
Compute: object java.lang.Object@d8355a8 on thread main
Emitting item on thread main
Listen on thread - main
Compute: object java.lang.Object@28d25987 on thread main
...

onSubscribe on 스레드 메인

만들기 on 스레드 - 메인

출처 방출 on 스레드 메인

방출 항목 on 스레드 메인

듣기 on 스레드 - 메인

계산: 객체 자바.lang.개체@d8355a8 on 스레드 메인

방출 항목 on 스레드 메인

듣기 on 스레드 - 메인

계산: 객체 자바.lang.개체@28d25987 on 스레드 메인

...

모든 것이 메인 스레드에서 연쇄적으로 일어나고 있습니다. 그 모든 노력에 비해 그다지 흥미롭지는 않습니다. 아직 RxJava의 모든 기능을 활용하지는 못했습니다. RxJava에 익숙하다면 코드를 비동기적으로 실행하는 것이 얼마나 쉬운지 알 수 있을 것입니다. 제가 예상하지 못했던 흥미로운 점이 있습니다. 다시 한 번 다른 포스팅에서 살펴볼 내용입니다.

해설: RxJava 배우기

처음 이 글을 쓰기로 마음먹었을 때는 RxJava의 내부를 공부하면서 배운 것들에 대해 이야기하고 싶었습니다. 결국 글 한 편에 담기에는 너무 많은 양이 되었습니다.

RxJava를 어렵게 만드는 요인 중 하나는 API의 수가 많다는 점이지만, 기본적인 것조차도 서로 얽혀 있는 인터페이스의 웹과 관련이 있습니다.

예를 들어, 다음과 같은 버전은 하나만 있습니다. Observable.create. . 옵저버블온구독 인스턴스를 인수로 사용할 수 있습니다. 충분히 간단합니다.

하지만 조금 더 자세히 살펴봅시다. 옵저버블온구독 는 하나의 메서드만 있는 인터페이스입니다, 구독. 다시 말하지만, 간단하지만 RxJava를 이해하기 까다로운 이유의 일부가 드러나기 시작합니다.

더 자세히 살펴보면 다음과 같습니다. 옵저버블온서브스크립트 구독 는 관찰 가능 이미터 를 인수로 사용할 수 있습니다. An 관찰 가능 이미터 를 확장합니다. 이미터를 관리하기 위한 몇 가지 메서드를 추가하고 일회용. An 이미터와 거의 동일한 인터페이스를 가지고 있는 것으로 밝혀졌습니다. 관찰 가능. . onSubscribe 메서드를 사용합니다.

대부분 인터페이스 정의가 있습니다. 구현에 대해서는 아직 다루지 않았으며 여기서는 표면적인 부분만 다루고 있습니다.

코드는 매우 흥미롭습니다. 자세히 살펴보면 오퍼레이터가 전체 호출 체인에서 일어나는 일을 수정하는 방법, RxJava가 버퍼링을 수행하는 방법, 일부 오퍼레이션이 전달할 항목 수를 알리는 방법 등에 대해 알아볼 수 있습니다.

앞으로 더 많은 것을 탐구하고 배운 것을 전달할 수 있기를 기대합니다.

포스트 스크립트

Couchbase는 오픈 소스이며 무료 체험.
시작하기 와 함께 샘플 코드, 예제 쿼리, 튜토리얼 등.
더 많은 리소스 찾기 개발자 포털.
트위터에서 팔로우하기 카우치베이스 개발.
질문에 대한 답변을 게시할 수 있습니다. 포럼.
당사는 다음 사항에 적극적으로 참여합니다. 스택 오버플로.
질문, 의견, 보고 싶은 주제 등이 있으면 트위터에서 저에게 연락해 주세요. 호드그릴리

ReactiveX 로고 사용 리액티브X 프로젝트 조건에 따라 크리에이티브 커먼즈 저작자표시 3.0 라이선스.

호드 그릴리, 개발자 옹호자, 카우치베이스

Platform

Self-Managed

Services

Capabilities

Why Couchbase?

Migrate to Capella

By Use Case

By Industry

By Application Need

Popular Docs

By Developer Role

Quickstart

Resource Center

About

Partnerships

Our Services

Partners: Register a Deal

Ready to register a deal with Couchbase?

Marriott

모든 게시물

RxJava 살펴보기 - 리스너 콜백 래핑하기

소개

목표

시작하기

소스 만들기

리스팅: BasicSource.java

리스팅: UnboundSource.java

옵저버블로 전환

목록: Observables.java

인스턴스화 및 구독

출력

해설: RxJava 배우기

포스트 스크립트

작성자

게시자 호드 그릴리, 개발자 옹호자, 카우치베이스

댓글 남기기 응답 취소