Android 架构师之路20 响应式编程RxJava操作符源码分析与实现

操作符简介

操作符：将发出的数据进行处理并再发送
变化传播--通过操作符实现变化，并能向下传播

1.RxJava1操作符源码分析

1.Func1接口
2.Operator接口

1.1RxJava1实例

                Observable.create(new Observable.OnSubscribe<String>() {
                            @Override
                            public void call(Subscriber<? super String> subscriber) {
                                if (!subscriber.isUnsubscribed()) {
                                    subscriber.onNext("1");
                                    subscriber.onNext("2");
                                    subscriber.onCompleted();
                                }
                            }
                        }).
                        //处理
                        map(new Func1<String, Integer>() {
                            @Override
                            public Integer call(String s) {
                                return Integer.parseInt(s)+2;
                            }
                        }).
                        subscribe(new Observer<Integer>() {
                            @Override
                            public void onCompleted() {
                                Log.d("kpioneer", "onCompleted:");
                            }

                            @Override
                            public void onError(Throwable e) {

                            }

                            @Override
                            public void onNext(Integer integer) {
                                Log.d("kpioneer", "onNext:" + integer + ",integer instanceOf" + integer.getClass());
                            }
                        });

运行

06-11 10:10:20.008 14148-14148/com.haocai.rxjavademo D/kpioneer: onNext:3,integer instanceOfclass java.lang.Integer
06-11 10:10:20.008 14148-14148/com.haocai.rxjavademo D/kpioneer: onNext:4,integer instanceOfclass java.lang.Integer
06-11 10:10:20.008 14148-14148/com.haocai.rxjavademo D/kpioneer: onCompleted:

1.2RxJava1操作符源码

RxJava1中OnSubscribeMap类

public final class OnSubscribeMap<T, R> implements OnSubscribe<R> {

    final Observable<T> source;

    final Func1<? super T, ? extends R> transformer;

    public OnSubscribeMap(Observable<T> source, Func1<? super T, ? extends R> transformer) {
        this.source = source;
        this.transformer = transformer;
    }

    @Override
    public void call(final Subscriber<? super R> o) {
        MapSubscriber<T, R> parent = new MapSubscriber<T, R>(o, transformer);
        o.add(parent);
        source.unsafeSubscribe(parent);
    }

    static final class MapSubscriber<T, R> extends Subscriber<T> {

        final Subscriber<? super R> actual;

        final Func1<? super T, ? extends R> mapper;

        boolean done;

        public MapSubscriber(Subscriber<? super R> actual, Func1<? super T, ? extends R> mapper) {
            this.actual = actual;
            this.mapper = mapper;
        }

        @Override
        public void onNext(T t) {
            R result;

            try {
                result = mapper.call(t);
            } catch (Throwable ex) {
                Exceptions.throwIfFatal(ex);
                unsubscribe();
                onError(OnErrorThrowable.addValueAsLastCause(ex, t));
                return;
            }

            actual.onNext(result);
        }

        @Override
        public void onError(Throwable e) {
            if (done) {
                RxJavaHooks.onError(e);
                return;
            }
            done = true;

            actual.onError(e);
        }


        @Override
        public void onCompleted() {
            if (done) {
                return;
            }
            actual.onCompleted();
        }

        @Override
        public void setProducer(Producer p) {
            actual.setProducer(p);
        }
    }

}

RxJava1中OnSubscribeLift类

public final class OnSubscribeLift<T, R> implements OnSubscribe<R> {

    final OnSubscribe<T> parent;

    final Operator<? extends R, ? super T> operator;

    public OnSubscribeLift(OnSubscribe<T> parent, Operator<? extends R, ? super T> operator) {
        this.parent = parent;
        this.operator = operator;
    }

    @Override
    public void call(Subscriber<? super R> o) {
        try {
            Subscriber<? super T> st = RxJavaHooks.onObservableLift(operator).call(o);
            try {
                // new Subscriber created and being subscribed with so 'onStart' it
                st.onStart();
                parent.call(st);
            } catch (Throwable e) {
                // localized capture of errors rather than it skipping all operators
                // and ending up in the try/catch of the subscribe method which then
                // prevents onErrorResumeNext and other similar approaches to error handling
                Exceptions.throwIfFatal(e);
                st.onError(e);
            }
        } catch (Throwable e) {
            Exceptions.throwIfFatal(e);
            // if the lift function failed all we can do is pass the error to the final Subscriber
            // as we don't have the operator available to us
            o.onError(e);
        }
    }
}

1.3变换的原理（核心操作符lift）:

1.接收原OnSubscribe的当前的Operator
2.创建一个新的OnSubscribe并返回新的Observable

    public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
        return unsafeCreate(new OnSubscribeLift<T, R>(onSubscribe, operator));
    }

3.用新的Subscriber包裹旧的Subscriber

        MapSubscriber<T, R> parent = new MapSubscriber<T, R>(o, transformer);
        o.add(parent);
        source.unsafeSubscribe(parent);

4.在新的Subscriber里做完变换再传给旧的Subscriber

        @Override
        public void onNext(T t) {
            R result;

            try {
                result = mapper.call(t);
            } catch (Throwable ex) {
                Exceptions.throwIfFatal(ex);
                unsubscribe();
                onError(OnErrorThrowable.addValueAsLastCause(ex, t));
                return;
            }

            actual.onNext(result);
        }

    public final <R> Observable<R> lift(final Operator<? extends R, ? super T> operator) {
        return unsafeCreate(new OnSubscribeLift<T, R>(onSubscribe, operator));
    }

分析

核心实现使用了代理机制

2.RxJava2操作符源码分析

2.1.RxJava2实例

                Observable.
                        create(new ObservableOnSubscribe<String>() {
                            @Override
                            public void subscribe(ObservableEmitter<String> e) throws Exception {
                                if (!e.isDisposed()) {
                                    e.onNext("1");
                                    e.onNext("2");
                                    e.onComplete();
                                }
                            }
                        }).
                        map(new Function<String, Integer>() {
                            @Override
                            public Integer apply(String s) throws Exception {
                                return Integer.parseInt(s)+2;
                            }
                        }).
                        subscribe(new Observer<Integer>() {
                            @Override
                            public void onSubscribe(Disposable d) {
                                Log.d("kpioneer", "onSubscribe:");
                            }

                            @Override
                            public void onNext(Integer value) {
                                Log.d("kpioneer", "onNext:" + value);
                            }

                            @Override
                            public void onError(Throwable e) {

                            }

                            @Override
                            public void onComplete() {
                                Log.d("kpioneer", "onComplete" );
                            }
                        });

                Flowable.
                        create(new FlowableOnSubscribe<String>() {
                            @Override
                            public void subscribe(FlowableEmitter<String> e) throws Exception {
                                if (!e.isCancelled()) {
                                    e.onNext("1");
                                    e.onNext("2");
                                    e.onComplete();
                                }
                            }
                        }, BackpressureStrategy.DROP).
                        map(new Function<String, Integer>() {
                            @Override
                            public Integer apply(String s) throws Exception {
                                return Integer.parseInt(s)+2;
                            }
                        }).
                        subscribe(new Subscriber<Integer>() {
                            @Override
                            public void onSubscribe(Subscription s) {
                                s.request(Long.MAX_VALUE);
                                Log.d("kpioneer", "onSubscribe");
                            }

                            @Override
                            public void onNext(Integer integer) {
                                Log.d("kpioneer", "onNext:" + integer);
                            }

                            @Override
                            public void onError(Throwable t) {

                            }

                            @Override
                            public void onComplete() {
                                Log.d("kpioneer", "onComplete");
                            }
                        });

运行

06-11 10:20:56.688 16675-16675/com.haocai.rxjavademo D/kpioneer: onSubscribe:
06-11 10:20:56.688 16675-16675/com.haocai.rxjavademo D/kpioneer: onNext:3
06-11 10:20:56.698 16675-16675/com.haocai.rxjavademo D/kpioneer: onNext:4
06-11 10:20:56.698 16675-16675/com.haocai.rxjavademo D/kpioneer: onComplete
06-11 10:20:56.758 16675-16675/com.haocai.rxjavademo D/kpioneer: onSubscribe
06-11 10:20:56.758 16675-16675/com.haocai.rxjavademo D/kpioneer: onNext:3
06-11 10:20:56.758 16675-16675/com.haocai.rxjavademo D/kpioneer: onNext:4
06-11 10:20:56.768 16675-16675/com.haocai.rxjavademo D/kpioneer: onComplete

2.2.RxJava2操作符源码

Function接口

ObservableMap ：无背压

public final class ObservableMap<T, U> extends AbstractObservableWithUpstream<T, U> {
    final Function<? super T, ? extends U> function;

    public ObservableMap(ObservableSource<T> source, Function<? super T, ? extends U> function) {
        super(source);
        this.function = function;
    }

    @Override
    public void subscribeActual(Observer<? super U> t) {
        source.subscribe(new MapObserver<T, U>(t, function));
    }


    static final class MapObserver<T, U> extends BasicFuseableObserver<T, U> {
        final Function<? super T, ? extends U> mapper;

        MapObserver(Observer<? super U> actual, Function<? super T, ? extends U> mapper) {
            super(actual);
            this.mapper = mapper;
        }

        @Override
        public void onNext(T t) {
            if (done) {
                return;
            }

            if (sourceMode != NONE) {
                actual.onNext(null);
                return;
            }

            U v;

            try {
                v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
            } catch (Throwable ex) {
                fail(ex);
                return;
            }
            actual.onNext(v);
        }

        @Override
        public int requestFusion(int mode) {
            return transitiveBoundaryFusion(mode);
        }

        @Nullable
        @Override
        public U poll() throws Exception {
            T t = qs.poll();
            return t != null ? ObjectHelper.<U>requireNonNull(mapper.apply(t), "The mapper function returned a null value.") : null;
        }
    }
}

1.ObservableMap继承了AbstractObservableWithUpstream抽象类
2.利用了AbstractObservableWithUpstream中subscribeActual方法
3.用原Observable去subscribe变换后的Observer

public final class ObservableLift<R, T> extends AbstractObservableWithUpstream<T, R> {
    /** The actual operator. */
    final ObservableOperator<? extends R, ? super T> operator;

    public ObservableLift(ObservableSource<T> source, ObservableOperator<? extends R, ? super T> operator) {
        super(source);
        this.operator = operator;
    }

    @Override
    public void subscribeActual(Observer<? super R> s) {
        Observer<? super T> observer;
        try {
            observer = ObjectHelper.requireNonNull(operator.apply(s), "Operator " + operator + " returned a null Observer");
        } catch (NullPointerException e) { // NOPMD
            throw e;
        } catch (Throwable e) {
            Exceptions.throwIfFatal(e);
            // can't call onError because no way to know if a Disposable has been set or not
            // can't call onSubscribe because the call might have set a Disposable already
            RxJavaPlugins.onError(e);

            NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
            npe.initCause(e);
            throw npe;
        }

        source.subscribe(observer);
    }
}

FlowableMap：有背压

public final class FlowableMap<T, U> extends AbstractFlowableWithUpstream<T, U> {
    final Function<? super T, ? extends U> mapper;
    public FlowableMap(Flowable<T> source, Function<? super T, ? extends U> mapper) {
        super(source);
        this.mapper = mapper;
    }

    @Override
    protected void subscribeActual(Subscriber<? super U> s) {
        if (s instanceof ConditionalSubscriber) {
            source.subscribe(new MapConditionalSubscriber<T, U>((ConditionalSubscriber<? super U>)s, mapper));
        } else {
            source.subscribe(new MapSubscriber<T, U>(s, mapper));
        }
    }

    static final class MapSubscriber<T, U> extends BasicFuseableSubscriber<T, U> {
        final Function<? super T, ? extends U> mapper;

        MapSubscriber(Subscriber<? super U> actual, Function<? super T, ? extends U> mapper) {
            super(actual);
            this.mapper = mapper;
        }

        @Override
        public void onNext(T t) {
            if (done) {
                return;
            }

            if (sourceMode != NONE) {
                actual.onNext(null);
                return;
            }

            U v;

            try {
                v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
            } catch (Throwable ex) {
                fail(ex);
                return;
            }
            actual.onNext(v);
        }

        @Override
        public int requestFusion(int mode) {
            return transitiveBoundaryFusion(mode);
        }

        @Nullable
        @Override
        public U poll() throws Exception {
            T t = qs.poll();
            return t != null ? ObjectHelper.<U>requireNonNull(mapper.apply(t), "The mapper function returned a null value.") : null;
        }
    }

    static final class MapConditionalSubscriber<T, U> extends BasicFuseableConditionalSubscriber<T, U> {
        final Function<? super T, ? extends U> mapper;

        MapConditionalSubscriber(ConditionalSubscriber<? super U> actual, Function<? super T, ? extends U> function) {
            super(actual);
            this.mapper = function;
        }

        @Override
        public void onNext(T t) {
            if (done) {
                return;
            }

            if (sourceMode != NONE) {
                actual.onNext(null);
                return;
            }

            U v;

            try {
                v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
            } catch (Throwable ex) {
                fail(ex);
                return;
            }
            actual.onNext(v);
        }

        @Override
        public boolean tryOnNext(T t) {
            if (done) {
                return false;
            }

            U v;

            try {
                v = ObjectHelper.requireNonNull(mapper.apply(t), "The mapper function returned a null value.");
            } catch (Throwable ex) {
                fail(ex);
                return true;
            }
            return actual.tryOnNext(v);
        }

        @Override
        public int requestFusion(int mode) {
            return transitiveBoundaryFusion(mode);
        }

        @Nullable
        @Override
        public U poll() throws Exception {
            T t = qs.poll();
            return t != null ? ObjectHelper.<U>requireNonNull(mapper.apply(t), "The mapper function returned a null value.") : null;
        }
    }

1.FlowableMap继承了AbstractFlowableWithUpstream
2.利用了AbstractFlowableWithUpstream中的subscribeActual方法
3.用原Flowable去subscribe变换后的Subscriber
FlowableLift

public final class FlowableLift<R, T> extends AbstractFlowableWithUpstream<T, R> {
    /** The actual operator. */
    final FlowableOperator<? extends R, ? super T> operator;

    public FlowableLift(Flowable<T> source, FlowableOperator<? extends R, ? super T> operator) {
        super(source);
        this.operator = operator;
    }

    @Override
    public void subscribeActual(Subscriber<? super R> s) {
        try {
            Subscriber<? super T> st = operator.apply(s);

            if (st == null) {
                throw new NullPointerException("Operator " + operator + " returned a null Subscriber");
            }

            source.subscribe(st);
        } catch (NullPointerException e) { // NOPMD
            throw e;
        } catch (Throwable e) {
            Exceptions.throwIfFatal(e);
            // can't call onError because no way to know if a Subscription has been set or not
            // can't call onSubscribe because the call might have set a Subscription already
            RxJavaPlugins.onError(e);

            NullPointerException npe = new NullPointerException("Actually not, but can't throw other exceptions due to RS");
            npe.initCause(e);
            throw npe;
        }
    }
}

2.3Operator接口

1.实现此接口
2.在subscribeActual中做变换
3.用于扩展自定义操作符

分析

RxJava2 有背压和无背压核心实现使用了代理机制

3.RxJava1 操作符功能仿写实现

Operator接口实现

Operator接口是操作符的抽象接口
个操作符实现Operator接口用于处理具体的变换

lift操作符

变换的基本原理
各操作符均实现Operator接口，并调用lift操作符

map操作符

最基本的操作符
顾名思义，用于做映射

public class Caller<T> {
    final OnCall<T> onCall;

    public Caller(OnCall<T> onCall) {
        this.onCall = onCall;
    }

    public static <T> Caller<T> create(OnCall<T> onCall) {
        return new Caller<>(onCall);
    }

    public Calling call(Receiver<T> receiver) {
        this.onCall.call(receiver);
        return receiver;
    }

    public final <R> Caller<R> lift(final Operator<R, T> operator) {
        return create(new OnCallLift<>(onCall, operator));
    }

    public final <R> Caller<R> map(Func1<T, R> func) {
        return lift(new MapOperator<T, R>(func));
    }

    public interface OnCall<T> extends Action1<Receiver<T>> {

    }

    public interface Operator<R, T> extends Func1<Receiver<R>, Receiver<T>> {

    }
}

public interface Func1<T,R>{
    R call(T t);
}

public class MapOperator<T, R> implements Caller.Operator<R, T> {


    private final Func1<T, R> mapper;

    public MapOperator(Func1<T, R> mapper) {
        this.mapper = mapper;
    }

    @Override
    public Receiver<T> call(Receiver<R> rReceiver) {
        return new MapReceiver<>(rReceiver, this.mapper);
    }
}

public class MapReceiver<T, R> extends Receiver<T> {

    private final Receiver<R> actual;


    private final Func1<T, R> mapper;

    public MapReceiver(Receiver<R> actual, Func1<T, R> mapper) {
        this.actual = actual;
        this.mapper = mapper;
    }

    @Override
    public void onCompleted() {
        this.actual.onCompleted();
    }

    @Override
    public void onError(Throwable t) {
        this.actual.onError(t);
    }

    @Override
    public void onReceive(T t) {
        R tR = this.mapper.call(t);
        this.actual.onReceive(tR);
    }
}

public class OnCallLift<T, R> implements Caller.OnCall<R> {

    private final Caller.OnCall<T> parent;

    private final Caller.Operator<R, T> operator;

    public OnCallLift(Caller.OnCall<T> parent, Caller.Operator<R, T> operator) {
        this.parent = parent;
        this.operator = operator;
    }

    @Override
    public void call(Receiver<R> rReceiver) {
        Receiver<T> tReceiver = this.operator.call(rReceiver);
        this.parent.call(tReceiver);
    }

}

调用

public class Lesson2_2Activity extends AppCompatActivity {
    @Override
    protected void onCreate(final Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_custom_test);
        ButterKnife.bind(this);

    }

    @OnClick(R.id.testDo)
    public void onViewClicked() {
        Caller.
                create(new Caller.OnCall<String>() {
                    @Override
                    public void call(Receiver<String> stringReceiver) {
                        if (!stringReceiver.isUnCalled()) {
                            stringReceiver.onReceive("1");
                            stringReceiver.onReceive("2");
                            stringReceiver.onCompleted();
                        }
                    }
                }).
                map(new Func1<String, Integer>() {
                    @Override
                    public Integer call(String s) {
                        return Integer.parseInt(s)+2;
                    }
                }).
                call(new Receiver<Integer>() {
                    @Override
                    public void onCompleted() {

                        Log.d("kpioneer", "onCompleted");
                    }

                    @Override
                    public void onError(Throwable t) {

                    }

                    @Override
                    public void onReceive(Integer integer) {
                        Log.d("kpioneer", "onReceive:" + integer);
                    }
                });
    }
}

Log输出

06-11 16:45:26.988 10850-10850/com.haocai.rxjavademo D/kpioneer: onReceive:3
06-11 16:45:26.988 10850-10850/com.haocai.rxjavademo D/kpioneer: onReceive:4
06-11 16:45:26.988 10850-10850/com.haocai.rxjavademo D/kpioneer: onCompleted

3.RxJava2(无背压) 操作符功能仿写实现

CallerWithUpstream(类似于AbstractObservableWithUpstream)

一个抽象类
有callActual方法
实现操作符需实现此方法

map操作符

最基本的操作符
顾名思义，用于做映射

CallerOperator接口

在callActual中做变换
可用于扩展操作符

4.RxJava2(有背压) 操作符功能仿写实现

TelephonerOperator接口

在callActual中做变换
可用于扩展操作符

TelephonerWithUpstream(类似于AbstractObservableWithUpstream)

一个抽象类
有callActual方法
实现操作符需实现此方法

源码下载

https://github.com/kpioneer123/RxJavaLearning

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 213,186评论 6赞 492
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 90,858评论 3赞 387
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 158,620评论 0赞 348
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 56,888评论 1赞 285
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 66,009评论 6赞 385
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 50,149评论 1赞 291
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 39,204评论 3赞 412
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 37,956评论 0赞 268
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 44,385评论 1赞 303
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 36,698评论 2赞 327
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 38,863评论 1赞 341
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 34,544评论 4赞 335
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 40,185评论 3赞 317
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 30,899评论 0赞 21
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 32,141评论 1赞 267
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 46,684评论 2赞 362
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 43,750评论 2赞 351

Android 架构师之路20 响应式编程RxJava操作符源码分析与实现

Android 架构师之路20 响应式编程RxJava操作符源码分析与实现

操作符简介

1.RxJava1操作符源码分析

1.1RxJava1实例

运行

1.2RxJava1操作符源码

RxJava1中OnSubscribeMap类

RxJava1中OnSubscribeLift类

1.3变换的原理（核心操作符lift）:

分析

2.RxJava2操作符源码分析

2.1.RxJava2实例

运行

2.2.RxJava2操作符源码

ObservableMap ：无背压

FlowableMap：有背压

2.3Operator接口

分析

3.RxJava1 操作符功能仿写实现

Operator接口实现

lift操作符

map操作符

调用

Log输出

3.RxJava2(无背压) 操作符功能仿写实现

CallerWithUpstream(类似于AbstractObservableWithUpstream)

map操作符

CallerOperator接口

相关代码：

4.RxJava2(有背压) 操作符功能仿写实现

TelephonerOperator接口

TelephonerWithUpstream(类似于AbstractObservableWithUpstream)

相关源码

源码下载

推荐阅读更多精彩内容

Android 架构师之路20 响应式编程RxJava操作符源码分析与实现

操作符简介

1.RxJava1操作符源码分析

1.1RxJava1实例

运行

1.2RxJava1操作符源码

RxJava1中OnSubscribeMap类

RxJava1中OnSubscribeLift类

1.3变换的原理（核心操作符lift）:

分析

2.RxJava2操作符源码分析

2.1.RxJava2实例

运行

2.2.RxJava2操作符源码

ObservableMap ：无背压

FlowableMap： 有背压

2.3Operator接口

分析

3.RxJava1 操作符功能仿写实现

Operator接口实现

lift操作符

map操作符

调用

Log输出

3.RxJava2(无背压) 操作符功能仿写实现

CallerWithUpstream(类似于AbstractObservableWithUpstream)

map操作符

CallerOperator接口

相关代码：

4.RxJava2(有背压) 操作符功能仿写实现

TelephonerOperator接口

TelephonerWithUpstream(类似于AbstractObservableWithUpstream)

相关源码

源码下载

推荐阅读更多精彩内容

FlowableMap：有背压