RX에 맞게 코드가 완료되었습니다. onSubscribe 스레드를 차단하지 않습니다.

Question

새 스레드를 생성하여 네트워킹 작업을 수행하는 기존의 RX 코드가 아닙니다. 작업이 완료되면 콜백에서 하나의 메소드를 호출합니다.

이 코드가 실행되는 스레드를 제어 할 권한이 없습니다. 그것은 유산이며 새로운 Thread을 단독으로 생성합니다.

interface Callback { 
    void onSuccess(); 
} 

static void executeRequest(String name, Callback callback) { 
    new Thread(() -> { 
     try { 
      System.out.println(" Starting... " + name); 
      Thread.sleep(2000); 
      System.out.println(" Finishing... " + name); 
      callback.onSuccess(); 
     } catch (InterruptedException ignored) {} 
    }).start(); 
} 

은 내가 RX Completable이를 변환 할 :

이

은 같은 단순화 할 수있다. 이렇게하려면 Completable#create()을 사용합니다. CompletableEmitter의 구현은 executeRequest을 전달하며 요청이 완료되면 Callback 신호를 전달합니다.

또한 디버깅을 돕기 위해 가입 할 때 로그 추적을 인쇄합니다.

static Completable createRequestCompletable(String name) { 
     return Completable.create(e -> executeRequest(name, e::onComplete)) 
       .doOnSubscribe(d -> System.out.println("Subscribed to " + name)); 
} 

예상대로 작동합니다. Completable은 "요청"이 완료되고 콜백이 호출 된 후에 만 ​​완료됩니다.

trampoline 스케줄러에서 이러한 완료를 구독 할 때 두 번째 요청을 구독하기 전에 첫 번째 요청이 완료 될 때까지 기다리지 않는 것이 문제입니다.

이 코드 :

final Completable c1 = createRequestCompletable("1"); 
c1.subscribeOn(Schedulers.trampoline()).subscribe(); 

final Completable c2 = createRequestCompletable("2"); 
c2.subscribeOn(Schedulers.trampoline()).subscribe(); 

출력은 : 보시다시피

Subscribed to 1 
    Starting... 1 
Subscribed to 2 
    Starting... 2 
    Finishing... 1 
    Finishing... 2 

, 내가 trampoline에 가입하고있어 경우에도 완료 처음 Completable 전에 두 번째 Completable에 등록합니다.

내가 출력, 최초의 초 대기가 완료되도록 completables을 대기하고 싶습니다이 :

Subscribed to 1 
    Starting... 1 
    Finishing... 1 
Subscribed to 2 
    Starting... 2 
    Finishing... 2 

나는 노동자에서 수행되는 문제는 업무와 관련이 확신 실. Completable의 구현이 새 스레드를 생성하지 않으면 예상대로 작동합니다. 하지만 이것은 레거시 코드이며 수정을 시도하지 않고 RX에 적용하는 것입니다.

참고 : 요청은 프로그램의 다른 지점에서 실행됩니다. andThen 또는 concat을 사용하여 직렬화 된 실행을 구현할 수 없습니다.

Answer 1

Latch으로 서브 스크립 션 Thread을 명시 적으로 차단하여 Completables을 순차적으로 실행했습니다. 그러나 이것이 RX에서이를 수행하는 관용적 인 방법이라고 생각하지 않습니다. 왜 이렇게해야하는지 이해하지 못하고 Completable이 완료 될 때까지 스레드가 차단되지 않습니다.

static Completable createRequestCompletable(String name) { 
    final CountDownLatch latch = new CountDownLatch(1); 
    return Completable.create(e -> { 
     executeRequest(name,() -> { 
      e.onComplete(); 
      latch.countDown(); 
     }); 
     latch.await(); 
    }) 
    .doOnSubscribe(disposable -> System.out.println("Subscribed to " + name)); 
}