-- “通过OnSubscribe的源码的注释：意思是当Observable被订阅(subscribe) OnSubscribe接口的call方法会被执行。”

--subscriber.onNext(""+i) 这里的i转换为String 后作为参数传递为subscribe(new Action1<String>))中的String

create操作符的基本使用

顾名思义，Create操作符是用来创建一个Observable的。下面来看一个简单的代码段：

Observable.create(new Observable.OnSubscribe<String>() {@Overridepublic void call(Subscriber<? super String> subscriber) {//Emit Data}
})

create方法接收一个参数Observable.OnSubscribe 来看下它的源码：

    /*** Invoked when Observable.subscribe is called.*/public interface OnSubscribe<T> extends Action1<Subscriber<? super T>> {// cover for generics insanity}

Observable.OnSubscribe 说白了就是一个继承了Action1接口的接口：

public interface Action1<T> extends Action {void call(T t);
}

/*** All Action interfaces extend from this.* <p>* Marker interface to allow instanceof checks.*/
public interface Action extends Function {}

/*** All Func and Action interfaces extend from this.* <p>* Marker interface to allow instanceof checks.*/
public interface Function {}

它们的继承关系如下：
Observable.OnSubscribe <- Action1 <- Action <- Function

create()方法也就是个工厂方法：

public static <T> Observable<T> create(OnSubscribe<T> f) {return new Observable<T>(hook.onCreate(f));
}

通过OnSubscribe的源码的注释
Invoked when Observable.subscribe is called. 意思是当Observable被订阅(subscribe)
OnSubscribe接口的call方法会被执行。

知道如何创建(create)Observable, 接下来我们看下如何订阅它：

Observable.create(new Observable.OnSubscribe<String>() {@Overridepublic void call(Subscriber<? super String> subscriber) {for (int i = 0; i < 5; i++) {printLog(tvLogs, "Emit Data:", i + "");subscriber.onNext("" + i);}}
})
.subscribe(new Action1<String>() {@Overridepublic void call(String s) {//showToast(s);printLog(tvLogs, "Consume Data:", s);}
});

当调用了subscribe方法 Observable.OnSubscribe的call方法就会被执行，在Observable.OnSubscribe的call方法中循环了调用了5次subscriber.onNext，在subscribe的Action1回调就会接受5次回调。

Emit Data:'0' , Thread Name:RxCachedThreadScheduler-1
Emit Data:'1' , Thread Name:RxCachedThreadScheduler-1
Emit Data:'2' , Thread Name:RxCachedThreadScheduler-1
Emit Data:'3' , Thread Name:RxCachedThreadScheduler-1
Emit Data:'4' , Thread Name:RxCachedThreadScheduler-1
Consume Data:'0' , Thread Name:main
Consume Data:'1' , Thread Name:main
Consume Data:'2' , Thread Name:main
Consume Data:'3' , Thread Name:main
Consume Data:'4' , Thread Name:main

从输出的日志可以看到，我们还打印了Thread Name线程的名称，我们可以控制发送数据、消费数据所在的线程。

.observeOn(AndroidSchedulers.mainThread())
.subscribeOn(Schedulers.io())

subscribeOn 设置Observable的call方法所在的线程【生产数据】

observeOn 设置subscribe的call方法所在的线程【消费数据】

Observable.OnSubscribe 的理解相关推荐

拆轮子系列--RxJava理解(一)--Map解析
本系列文章如下: 拆轮子系列--RxJava前奏篇拆轮子系列--RxJava理解(一)--Map解析拆轮子系列--RxJava理解(二)--subscribeOn 拆轮子系列--RxJava理解( ...
理解RxJava线程模型
RxJava作为目前一款超火的框架,它便捷的线程切换一直被人们津津乐道,本文从源码的角度,来对RxJava的线程模型做一次深入理解.(注:本文的多处代码都并非原本的RxJava的源码,而是用来说明逻辑 ...
理解 RxJava 线程模型
http://blog.saymagic.cn/2016/08/20/understand-rxjava-threading-model.html 主题线程 RxJava RxJava作为目前一款超 ...
rxjava onCompleted 方法的理解
注意:经过测试,onComplted之后,观察者将收不到消息,取消订阅函数得到执行 private void test(){Observable.create(new Observable.OnSub ...
Observable常用知识点整理
文章目录 Observable 基础概述操作符创建操作 Create Defer Empty/Never/Throw From Interva just Range Repeat Start T ...
observable java_RxJava学习笔记（1）创建Observable
RxJava文档:https://github.com/ReactiveX/RxJava/wiki RxJava中文文档:https://mcxiaoke.gitbooks.io/rxdocs/con ...
android rxjava 回调,android – 使用回调/监听器链接RxJava observable
调整clemp6r的解决方案,这是另一个既不需要主题也不需要嵌套订阅的解决方案: api.getKey().flatMap(new Func1>() { @Override public Obs ...
RxJava 源码解析之观察者模式
了解 RxJava 的应该都知道是一个基于事务驱动的库,响应式编程的典范.提到事务驱动和响应就不得不说说,设计模式中观察者模式,已经了解的朋友,可以直接跳过观察者模式的介绍,直接到 RxJava 源码 ...
用大白话讲解RxJava原理
近日外媒报道称,Google 其应用商店 Play Store 一次下架了超过 600 个违规 App:其中中国.印度以及新加坡为本次大规模下架 App 开发者前三的国家,来自中国的猎豹移动所开发的 ...

Observable.OnSubscribe 的理解

create操作符的基本使用

Observable.OnSubscribe 的理解相关推荐

最新文章

热门文章