为何不建议使用Observable.create？RxJava替代方案与风险解析

Great question! I totally get why this feels confusing—most RxJava tutorials just drop the rule "avoid Observable.create" without breaking down exactly what makes it risky. Let's break this down with concrete examples, beyond just backpressure.

The core issue with Observable.create is that it puts all the responsibility of following RxJava's strict Observable contract on you. Factory methods like just, fromCallable, or range already handle these rules under the hood, so you don't have to worry about accidental bugs.

1. Manual Subscription & Cleanup Is Error-Prone #

When using create, you're responsible for stopping event emissions as soon as the observer cancels their subscription. Skip this step, and you'll end up with memory leaks, invalid events sent to disposed emitters, or wasted resources.

Bad Example: Unhandled Disposal #

Observable.create(emitter -> {
    // Spin up a background thread to emit events
    new Thread(() -> {
        for (int i = 0; i < 100; i++) {
            // No check if the observer has disposed the subscription
            emitter.onNext(i);
            try { Thread.sleep(100); } catch (InterruptedException e) {}
        }
        emitter.onComplete();
    }).start();
})
.subscribe(
    num -> System.out.println("Received: " + num),
    err -> err.printStackTrace()
)
.dispose(); // Cancel the subscription immediately

Even after we call dispose(), the background thread keeps running and emitting events. This will throw an IllegalStateException (since the emitter is disposed) and waste system resources.

Better Alternative: fromCallable #

Factory methods handle disposal automatically. If the subscription is canceled, they'll interrupt running tasks (where possible) and stop emissions:

Observable.fromCallable(() -> {
    for (int i = 0; i < 100; i++) {
        // Check if the thread was interrupted (triggered by dispose())
        if (Thread.currentThread().isInterrupted()) {
            throw new InterruptedException("Task canceled");
        }
        Thread.sleep(100);
    }
    return "Done";
})
.subscribe(
    result -> System.out.println("Result: " + result),
    err -> System.out.println("Error: " + err.getMessage())
)
.dispose();

Here, dispose() triggers a thread interrupt, and the task cleanly stops without sending invalid events.

2. It's Easy to Break the Observable Contract #

RxJava has non-negotiable rules for Observables:

• You can call onComplete() or onError() at most once.
• After onComplete()/onError(), no more onNext() calls are allowed.

With create, it's shockingly easy to violate these rules by accident.

Bad Example: Invalid Post-Error Emission #

Observable.create(emitter -> {
    try {
        emitter.onNext("First value");
        throw new RuntimeException("Oops, something broke");
    } catch (Exception e) {
        emitter.onError(e);
    }
    // Accidentally send another event after onError()
    emitter.onNext("Second value");
})
.subscribe(
    s -> System.out.println("Received: " + s),
    err -> System.out.println("Error: " + err.getMessage())
);

This code will throw an IllegalStateException because we called onNext() after onError(). Factory methods prevent this automatically: fromCallable will only emit a single value or an error, never both.

3. Thread Scheduling Traps #

Observable.create doesn't enforce any thread rules—events are emitted on the same thread that subscribed to the Observable. This can lead to accidental UI blocking (on Android) or unexpected concurrency bugs if you forget to specify schedulers.

Bad Example: Blocking the Main Thread #

// On Android, this would freeze the UI!
Observable.create(emitter -> {
    // Slow, blocking operation (e.g., reading a large file)
    String largeData = readHugeFileFromDisk();
    emitter.onNext(largeData);
    emitter.onComplete();
})
.subscribe(data -> updateUIWithData(data));

If you run this on the Android main thread, it'll block the UI until the file is read—something RxJava is supposed to help you avoid.

Better Alternative: fromCallable + Schedulers #

Factory methods play nicely with subscribeOn and observeOn to handle thread scheduling safely:

Observable.fromCallable(() -> readHugeFileFromDisk())
    .subscribeOn(Schedulers.io()) // Run the slow task on an IO thread
    .observeOn(AndroidSchedulers.mainThread()) // Deliver results to the UI thread
    .subscribe(data -> updateUIWithData(data));

This ensures the heavy work stays off the main thread, with zero manual thread management.

4. Backpressure Handling (The One You Already Mentioned) #

Backpressure is RxJava's way of balancing event emission speed with observer processing speed. Observable.create doesn't support backpressure by default—you have to manually implement the request(long n) logic to respect the observer's capacity, which is extremely complex.

Bad Example: Uncontrolled Event Flooding #

Observable.create(emitter -> {
    // Emit 10,000 events as fast as possible
    for (int i = 0; i < 10000; i++) {
        emitter.onNext(i);
    }
    emitter.onComplete();
})
.observeOn(Schedulers.single()) // Observer processes events slowly
.subscribe(
    num -> {
        Thread.sleep(10); // Simulate slow processing
        System.out.println("Processed: " + num);
    }
);

This will flood the observeOn queue with thousands of events, leading to high memory usage and potential OutOfMemoryErrors.

Better Alternative: range (Backpressure-Aware) #

Factory methods like range are built with backpressure support. They'll only emit events as fast as the observer can process them:

Observable.range(0, 10000)
    .observeOn(Schedulers.single())
    .subscribe(
        num -> {
            Thread.sleep(10);
            System.out.println("Processed: " + num);
        }
    );

Here, range respects the observer's request calls and emits events at a sustainable pace.

When Should You Use Observable.create? #

Only when you need a fully custom Observable that no factory method can handle—for example, wrapping a callback-based API like WebSockets or a custom sensor. Even then, you must:

• Always check emitter.isDisposed() before sending events.
• Never call onNext() after onComplete()/onError().
• Implement backpressure logic if your Observable emits many events.

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