一、描述

你可以使用 Create 操作符从头开始创建一个Observable，给这个操作符传递一个接受观察者作为参数的函数，编写这个函数让它的行为表现为一个Observable–恰当的调用观察者的onNext，onError和onCompleted方法。

一个形式正确的有限Observable必须尝试调用观察者的onCompleted正好一次或者它的onError正好一次，而且此后不能再调用观察者的任何其它方法。

RxJava将这个操作符实现为create方法。

建议你在传递给create方法的函数中检查观察者的isUnsubscribed状态，以便在没有观察者的时候，让你的Observable停止发射数据或者做昂贵的运算。

create方法默认不在任何特定的调度器上执行。

二、示意图

create

三、示例代码

Observable.create(new Observable.OnSubscribe<Integer>() {
    @Override
    public void call(Subscriber<? super Integer> observer) {
        try {
            if (!observer.isUnsubscribed()) {
                for (int i = 0; i < 4; i++) {
                    observer.onNext(i);
                }
                observer.onCompleted();
            }
        } catch (Exception e) {
            observer.onError(e);
        }
    }
}).subscribe(new Subscriber<Integer>() {
    @Override
    public void onCompleted() {
        System.out.println("onCompleted.");
    }

    @Override
    public void onError(Throwable e) {
        System.out.println("onError: " + e.getMessage());
    }

    @Override
    public void onNext(Integer integer) {
        System.out.println("onNext: " + integer);
    }
});

四、运行结果

onNext: 0
onNext: 1
onNext: 2
onNext: 3
onCompleted.

项目代码已上传到Github：https://github.com/SherlockShi/RxJavaBestPractise

五、参考资料

ReactiveX官方文档

ReactiveX文档中文翻译

PS：欢迎关注 SherlockShi 个人博客