RxJava->主子线程通信

  • 上一篇文章分析是如何创建线程, 其实仅仅就是把源码走了一遍, 里面很多细节由于目前本身能力有限无法继续分析, 待主子线程通信分析完以后把多线程, 并发包, 设计模式系统的过一遍再回过头把RxJava系列重新梳理一遍;

Example:

Observable
    .create(new ObservableOnSubscribe<Integer>() {
        @Override
        public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
            LogUtils.log(Note01.class, "subscribe()->ThreadName:" + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onComplete();
            }
        })
    .subscribeOn(Schedulers.newThread())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Observer<Integer>() {
        @Override
        public void onSubscribe(Disposable d) {
            LogUtils.log(Note01.class, "onSubscribe()->ThreadName:" + Thread.currentThread().getName());
        }

        @Override
        public void onNext(Integer value) {
            LogUtils.log(Note01.class, "onNext()->ThreadName:" + Thread.currentThread().getName());
        }

        @Override
        public void onError(Throwable e) {
            LogUtils.log(Note01.class, "onError()->ThreadName:" + Thread.currentThread().getName());
        }

        @Override
        public void onComplete() {
            LogUtils.log(Note01.class, "onComplete()->ThreadName:" + Thread.currentThread().getName());
        }
    });
  • 打印结果:
10-09 23:41:07.514 3178-3178/com.test onSubscribe()->ThreadName:main
10-09 23:41:07.514 3178-3572/com.test subscribe()->ThreadName:RxNewThreadScheduler-1
10-09 23:41:07.514 3178-3178/com.test onNext()->ThreadName:main
10-09 23:41:07.514 3178-3178/com.test onComplete()->ThreadName:main
  • 如何进行线程间通信:
.observeOn(AndroidSchedulers.mainThread());
public final class AndroidSchedulers {
    public static Scheduler mainThread() {
        return RxAndroidPlugins.onMainThreadScheduler(MAIN_THREAD);
    }
    private static final Scheduler MAIN_THREAD = RxAndroidPlugins.initMainThreadScheduler(
            new Callable<Scheduler>() {
                @Override public Scheduler call() throws Exception {
                    return MainHolder.DEFAULT;
                }
            });
    private static final class MainHolder {

        static final Scheduler DEFAULT = new HandlerScheduler(new Handler(Looper.getMainLooper()));
    }
}
final class HandlerScheduler extends Scheduler {
    private final Handler handler;

    HandlerScheduler(Handler handler) {
        this.handler = handler;
    }
}
  • 1、创建HandlerScheduler, 将引用赋给Scheduler;
  • 2、在主线程中创建Handler, HandlerScheduler内部持有Handler的引用;
public abstract class Observable<T> implements ObservableSource<T> {
    public final Observable<T> observeOn(Scheduler scheduler) {
        return observeOn(scheduler, false, bufferSize());
    }
    public final Observable<T> observeOn(Scheduler scheduler, boolean delayError, int bufferSize) {
        ObjectHelper.requireNonNull(scheduler, "scheduler is null");
        ObjectHelper.verifyPositive(bufferSize, "bufferSize");
        return RxJavaPlugins.onAssembly(new ObservableObserveOn<T>(this, scheduler, delayError, bufferSize));
    }
}
  • 又是老套路;
  • 1、Observable持有ObservableSubscribeOn的引用;
  • 2、ObservableObserveOn内部持有ObservableSubscribeOn的引用;
  • 3、ObservableSubscribeOn内部持有ObservableCreate的引用;

然后调用subscribe()时先从ObservableSubscribeOn开始:

public final class ObservableObserveOn<T> {
    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        Scheduler.Worker w = scheduler.createWorker();
        source.subscribe(new ObserveOnObserver<T>(observer, w, delayError, bufferSize));
    }
}
static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T> implements Observer<T>, Runnable {

    final Observer<? super T> actual;
    final Scheduler.Worker worker;
    SimpleQueue<T> queue;
    Disposable s;
    Throwable error;
    volatile boolean done;
    volatile boolean cancelled;
    int sourceMode;
    boolean outputFused;

    ObserveOnObserver(Observer<? super T> actual, Scheduler.Worker worker, boolean delayError, int bufferSize) {
        this.actual = actual;
        this.worker = worker;
    }
}
  • 然后递归到ObservableSubscribeOn中的subscribeActual()方法:
public final class ObservableSubscribeOn<T> extends AbstractObservableWithUpstream<T, T> {
    final Scheduler scheduler;

    public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
        super(source);
        this.scheduler = scheduler;
    }

    @Override
    public void subscribeActual(final Observer<? super T> s) {
        final SubscribeOnObserver<T> parent = new SubscribeOnObserver<T>(s);

        s.onSubscribe(parent);

        parent.setDisposable(scheduler.scheduleDirect(new Runnable() {
            @Override
            public void run() {
                source.subscribe(parent);
            }
        }));
    }
}
static final class SubscribeOnObserver<T> extends AtomicReference<Disposable> implements Observer<T>, Disposable {

    final Observer<? super T> actual;

    final AtomicReference<Disposable> s;

    SubscribeOnObserver(Observer<? super T> actual) {
        this.actual = actual;
        this.s = new AtomicReference<Disposable>();
    }

    @Override
    public void onSubscribe(Disposable s) {
        DisposableHelper.setOnce(this.s, s);
    }
}
  • 接下来又回到了上一节当中, 在子线程中调用source.subscribe()方法, source实际指向ObservableCreate.
public final class ObservableCreate<T> extends Observable<T> {
    final ObservableOnSubscribe<T> source;

    public ObservableCreate(ObservableOnSubscribe<T> source) {
        this.source = source;
    }

    @Override
    protected void subscribeActual(Observer<? super T> observer) {
        CreateEmitter<T> parent = new CreateEmitter<T>(observer);
        observer.onSubscribe(parent);
        source.subscribe(parent);
    }
}

static final class CreateEmitter<T> extends AtomicReference<Disposable> implements ObservableEmitter<T>, Disposable {

    final Observer<? super T> observer;

    CreateEmitter(Observer<? super T> observer) {
        this.observer = observer;
    }

    @Override
    public void onNext(T t) {
        if (!isDisposed()) {
            observer.onNext(t);
        }
    }

    @Override
    public void onComplete() {
        if (!isDisposed()) {
            try {
                observer.onComplete();
            } finally {
                dispose();
            }
        }
    }   
}
  • 1、将SubscribeOnObserver引用赋给CreateEmitter内部的Observer;
  • 2、将CreateEmitter引用赋给SubscribeOnObserver内部的Disposable;
  • 3、将CreateEmitter引用赋给ObservableOnSubscribe内部的ObservableEmitter;
  • 接下来通过发射器CreateEmitter调用onXXX()系列方法, 看看执行了那么逻辑: 通过上面分析知道CreateEmitter此时的onXXX()方法均在子线程中执行;
CreateEmitter.onNext() ->SubscribeOnObserver.onNext() -> ObserveOnObserver.onNext();

static final class ObserveOnObserver<T> {
    @Override
    public void onNext(T t) {
        schedule();
    }
    void schedule() {
        if (getAndIncrement() == 0) {
            worker.schedule(this);
        }
    }
}
  • Schedule指向HandlerScheduler:
public abstract static class Worker implements Disposable {
    public Disposable schedule(Runnable run) {
        return schedule(run, 0L, TimeUnit.NANOSECONDS);
    }
    public abstract Disposable schedule(Runnable run, long delay, TimeUnit unit);
}

final class HandlerScheduler extends Scheduler {
    @Override
    public Worker createWorker() {
        return new HandlerWorker(handler);
    }
}
private static final class HandlerWorker extends Worker {
    private final Handler handler;
    private volatile boolean disposed;
    HandlerWorker(Handler handler) {
        this.handler = handler;
    }

    @Override
    public Disposable schedule(Runnable run, long delay, TimeUnit unit) {
        if (disposed) {
            return Disposables.disposed();
        }
        run = RxJavaPlugins.onSchedule(run);
        ScheduledRunnable scheduled = new ScheduledRunnable(handler, run);
        Message message = Message.obtain(handler, scheduled);
        message.obj = this; // Used as token for batch disposal of this worker's runnables.
        handler.sendMessageDelayed(message, Math.max(0L, unit.toMillis(delay)));
        if (disposed) {
            handler.removeCallbacks(scheduled);
            return Disposables.disposed();
        }
        return scheduled;
    }
}
  • worker调用schedule-> handler.sendMessageDelayed(message, Math.max(0L, unit.toMillis(delay)))最终将ObserveOnObserver的run方法发送到主线程;
static final class ObserveOnObserver<T> extends BasicIntQueueDisposable<T> implements Observer<T>, Runnable {
    @Override
    public void run() {
        if (outputFused) {
            drainFused();
        } else {
            drainNormal();
        }
    }
}

顺便得出几个结果:

  • 1、要像观察者跑在主线程, 需要在方法被执行前调用.observeOn(AndroidSchedulers.mainThread()), 当遍历到该Observer时, 会通过handler.sendMessage实现主子线程通信;
  • 2、子线程的前提下想让某个方法跑在主线程, 则只需要在该方法前面调用.observeOn(AndroidSchedulers.mainThread())即可;
  • 3、连续多次调用scheduleOn(schedule.io), 则被观察者执行的操作所在线程为最后一次schedule.io创建的线程中;

目前来看, RxJava使用观察者的模式如下图流程(类型链表):

RxJava调用流程.png

每次调用类似subscribeOn(), doOnNext(), map()等方法时, 都会创建一个Observable子类, 然后该子类会包含前一个子类的引用, 并将自己的引用传给Observable下一个子类. 当源头ObservableEmitter调用onXXX()方法时, 会从最后一个子类开始调用, 然后递归调用pre.onXXX()方法直至Observable的第一个子类, 然后又会遍历从第一个子类调用next.onXXX()方法, 直至最后一个子类;

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

推荐阅读更多精彩内容

  • Rxjava
    我从去年开始使用 RxJava ，到现在一年多了。今年加入了 Flipboard 后，看到 Flipboard 的...
    Jason_andy阅读 5,460评论 7赞 62
  • RxJava 从入门到放弃再到不离不弃
    作者寄语 很久之前就想写一个专题，专写Android开发框架，专题的名字叫 XXX 从入门到放弃 ，沉淀了这么久，...
    戴定康阅读 7,616评论 13赞 85
  • 第一次写东西
    作为一个初高中时代作文从来只是在及格线徘徊的理科男，一直对文字抱着一种无所谓的态度。总觉得理科嘛，没必要对文字下功...
    层千阅读 254评论 0赞 2