RxJava2 Flowable zip异步调用问题：Java项目如何回到主线程

Hey there! Let's work through how to get your Flowable zip chain back to the original thread (your "main/current thread") in a plain Java project—no Android schedulers or hacky Thread.sleep() required.

First, let's recap your problem: you're using Flowable.zip() to run multiple methods (m1(), m2(), etc.) asynchronously on background threads, but you need the final combined result to be processed on the thread that initiated the subscription. Here are three practical solutions tailored for plain Java:

1. Use Schedulers.trampoline() for Simple Single-Threaded Scenarios #

The TrampolineScheduler is perfect if your "main thread" is a single, long-running thread (like a console app's main thread). It queues up downstream tasks to run on the current thread after any existing work finishes.

Example code:

import io.reactivex.Flowable;
import io.reactivex.schedulers.Schedulers;

public class RxJavaZipExample {
    public static void main(String[] args) throws InterruptedException {
        // Async execution of m1() and m2() on IO threads
        Flowable.zip(
                Flowable.fromCallable(RxJavaZipExample::m1).subscribeOn(Schedulers.io()),
                Flowable.fromCallable(RxJavaZipExample::m2).subscribeOn(Schedulers.io()),
                (result1, result2) -> String.format("Combined: %s + %s", result1, result2)
        )
        // Switch back to the thread that initiated the subscription (main thread here)
        .observeOn(Schedulers.trampoline())
        .subscribe(
                combinedResult -> {
                    // This runs on the main thread!
                    System.out.println("Result received on thread: " + Thread.currentThread().getName());
                    System.out.println("Final result: " + combinedResult);
                },
                Throwable::printStackTrace
        );

        // Keep the main thread alive (replace sleep with this to avoid timing issues)
        System.in.read();
    }

    private static String m1() {
        System.out.println("m1 running on thread: " + Thread.currentThread().getName());
        return "Result1";
    }

    private static String m2() {
        System.out.println("m2 running on thread: " + Thread.currentThread().getName());
        return "Result2";
    }
}

Why this works: #

• subscribeOn(Schedulers.io()) pushes m1() and m2() to background IO threads.
• observeOn(Schedulers.trampoline()) tells the downstream subscribe() callback to run on the original thread (main thread in this case).
• System.in.read() keeps the main thread alive so it can process the callback—no sleep() needed.

2. Custom Scheduler for Threads with Task Queues #

If your main thread uses a custom task queue (like a dedicated worker thread that processes jobs from a queue), create a custom scheduler using Schedulers.from() wrapped around an Executor that submits tasks back to your main thread's queue.

Example outline:

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Executor;

public class CustomMainThread {
    private final BlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<>();
    private final Thread mainThread;

    public CustomMainThread() {
        mainThread = new Thread(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                try {
                    // Process tasks from the queue sequentially
                    taskQueue.take().run();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        }, "Custom-Main-Thread");
        mainThread.start();
    }

    // Executor that submits tasks to the main thread's queue
    public Executor getMainThreadExecutor() {
        return taskQueue::offer;
    }
}

Then use it in your Flowable chain:

CustomMainThread customMain = new CustomMainThread();

Flowable.zip(
        Flowable.fromCallable(RxJavaZipExample::m1).subscribeOn(Schedulers.io()),
        Flowable.fromCallable(RxJavaZipExample::m2).subscribeOn(Schedulers.io()),
        (r1, r2) -> String.format("Combined: %s + %s", r1, r2)
)
.observeOn(Schedulers.from(customMain.getMainThreadExecutor()))
.subscribe(
        combinedResult -> {
            System.out.println("Result on custom main thread: " + Thread.currentThread().getName());
            System.out.println("Final result: " + combinedResult);
        },
        Throwable::printStackTrace
);

Why this works: #

Your custom Executor pushes the downstream callback into the main thread's task queue, which the main thread processes sequentially—perfect for scenarios where you have a dedicated application thread handling core logic (even in plain Java).

3. Blocking Get for Synchronous Result Retrieval #

If you can afford to block the original thread until the async tasks complete (e.g., in a script or simple utility), use blockingGet() to wait for the result and process it immediately on the original thread.

Example:

String combinedResult = Flowable.zip(
        Flowable.fromCallable(RxJavaZipExample::m1).subscribeOn(Schedulers.io()),
        Flowable.fromCallable(RxJavaZipExample::m2).subscribeOn(Schedulers.io()),
        (r1, r2) -> String.format("Combined: %s + %s", r1, r2)
)
.blockingGet();

// This runs directly on the original thread
System.out.println("Result on thread: " + Thread.currentThread().getName());
System.out.println("Final result: " + combinedResult);

Why this works: #

blockingGet() halts the current thread until the Flowable emits its final value, so the code after it runs directly on the original thread—no need for schedulers or thread management. Just note that this blocks the thread, so avoid it if your main thread needs to stay responsive.

内容的提问来源于stack exchange，提问作者user2870934