java - 适应RX Completable的代码不会阻塞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

如您所见，即使我订阅了 Completable，它也会在第一个 Completable完成之前订阅第二个 trampoline。

我想将 completables排队，以便第二个等待第一个完成，输出如下:

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

我确定问题与在工作线程中完成的工作有关。如果 Completable的实现未产生新线程，则它将按预期工作。
但这是遗留代码，我想做的是在不进行修改的情况下将其适应RX。

注意:请求在程序的不同位置执行-我不能使用 andThen或 concat来实现序列化的执行。

Answer 1

通过使用Completables显式阻止订阅Thread，我设法按顺序执行了Latch。
但是我不认为这是在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));
}