RxJava2.x是一个非常棒的流式编程,采用的观察者模式思想,事件的产生者产生事间之后发送给绑定的接受者,接受顺序与发送顺序一致.但是 是独立于RxJava1.x存在,本文讲解RxJava2.x的简

RxJava2 封装主要变化

compile 'io.reactivex.rxjava2:rxjava:2.1.0'
compile 'io.reactivex.rxjava2:rxandroid:2.0.1'

//观察者模式,这里产生事件,事件产生后发送给接受者,但是一定要记得将事件的产生者和接收者捆绑在一起,否则会出现错误
Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
//这里调用的方法会在产生事件之后会发送给接收者,接收者对应方法会收到
e.onNext("hahaha");
e.onError(new Exception("wulala"));
e.onComplete();
}}).subscribe(new Observer<String>() {
//接受者,根据事件产生者产生的事件调用不同方法
@Override
public void onSubscribe(Disposable d) {
Log.e(TAG, "onSubscribe: ");
}

@Override
public void onNext(String value) {
Log.e(TAG, "onNext: " + value);
}

@Override
public void onError(Throwable e) {
Log.e(TAG, "onError: ", e);
}

@Override
public void onComplete() {
Log.e(TAG, "onComplete: ");
}
});

subscribe();          //表示发送者随意发送数据,接受者什么都不管,什么都不接收.
subscribe(Consumer<? super T> onNext) {}     //只响应onNext()事件,其他的事件忽略.
subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError) {}         //含义同上
subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete) {}         //含义同上
subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe) {}     //含义同上

Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
e.onNext("hahaha");
e.onNext("hahaha");
e.onNext("hahaha");
Log.e(TAG,"运行在什么线程" + Thread.currentThread().getName());
e.onComplete();
}
}).subscribeOn(Schedulers.newThread())               //线程调度器,将发送者运行在子线程
.observeOn(AndroidSchedulers.mainThread())          //接受者运行在主线程
.subscribe(new Observer<String>() {
@Override
public void onSubscribe(Disposable d) {
Log.e(TAG, "onSubscribe: ");
Log.e(TAG,"接收在什么线程" + Thread.currentThread().getName());
}

@Override
public void onNext(String value) {
Log.e(TAG, "onNext: " + value);
}

@Override
public void onError(Throwable e) {
Log.e(TAG, "onError: ", e);
}

@Override
public void onComplete() {
Log.e(TAG, "onComplete: ");
}
});

- Schedulers.io()      io操作的线程, 通常io操作,如文件读写.
- Schedulers.computation()      计算线程,适合高计算,数据量高的操作.
- Schedulers.newThread()      创建一个新线程,适合子线程操作.
- AndroidSchedulers.mainThread()      Android的主线程,主线程

操作符之变换

Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
}
}).map(new Function<Integer, String>() {
@Override
public String apply(Integer integer) throws Exception {
return "我是变换过后的" + integer;
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.e("XYK",s);
}
});
}

读取一篇英文文章,将文章中的字符全部转换为大写.

//模拟一篇文章
String article = "fkjdsalijfofldaJFOIEjfldanlJR2OnfldajilwafkndaIUPO32,LFKjlijuJFLMA";
char[] chars = article.toCharArray();

StringBuffer sb = new StringBuffer();
for (int i = 0; i < chars.length; i++) {
Log.e(TAG,chars[i] + "");
if(chars[i] >= 'a' && chars[i] <= 'z'){
sb.append((chars[i] + "").toUpperCase());
}else{
sb.append(chars[i]);
}
}

Log.e(TAG,sb.toString());

//模拟一篇文章
String article = "fkjdsalijfofldaJFOIEjfldanlJR2OnfldajilwafkndaIUPO32,LFKjlijuJFLMA";

final char[] chars = article.toCharArray();
Observable.create(new ObservableOnSubscribe<Character>() {
@Override
public void subscribe(ObservableEmitter<Character> e) throws Exception {
for (int i = 0; i < chars.length; i++) {
e.onNext(chars[i]);
}
}
//delay  延时5秒发送
}).delay(5, TimeUnit.SECONDS)
//事件类型转换
.map(new Function<Character, String>() {
@Override
public String apply(Character s) throws Exception {
if (s >= 'a' && s <= 'z') {
return s.toString().toUpperCase();
} else {
return s.toString();
}
}
})
//线程调度
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.e(TAG, s);
}
});

Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
}
}).flatMap(new Function<Integer, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(Integer integer) throws Exception {
List<String> list = new ArrayList<String>();
for (int i = 0; i < 5; i++) {
list.add("我是变换过的" + integer);
}
return Observable.fromIterable(list);
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.e("XYK", s);
}
});
}

Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
}
}).concatMap(new Function<Integer, ObservableSource<String>>() {
@Override
public ObservableSource<String> apply(Integer integer) throws Exception {
List<String> list = new ArrayList<String>();
for (int i = 0; i < 5; i++) {
list.add("我是变换过的" + integer);
}
return Observable.fromIterable(list);
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.e("XYK", s);
}
});
}

过滤操作符

Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
for (int i = 0; i < 10000; i++) {
e.onNext(i);
}
}
}).observeOn(Schedulers.io())
.subscribeOn(AndroidSchedulers.mainThread())
.filter(new Predicate<Integer>() {
@Override
public boolean test(Integer integer) throws Exception {
return integer % 7 == 0;
}
}).subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e("XYK",integer + "");
}
});
}

Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
for (int i = 0; ; i++) {
e.onNext(i);
}
}
}).sample(1,TimeUnit.SECONDS)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e("XYK",integer + "");
}
});
}

Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
for (int i = 0; ; i++) {
e.onNext(i);
}
}
}).take(3)
//.takeList(3)
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e("XYK",integer + "");
}
});
}

Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
for (int i = 0;i < 50; i++) {
e.onNext(i);
}
}
}).take(3)
//生成重复事件
.repeat(3)
.distinct()
.subscribe(new Consumer<Integer>() {
@Override
public void accept(Integer integer) throws Exception {
Log.e("XYK",integer + "");
}
});

组合操作符

Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
e.onNext(1);
e.onNext(2);
e.onNext(3);
e.onNext(4);
}
});Observable<String> observable1 = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
e.onNext("这是");
e.onNext("这个是");
e.onNext("这个则是");
}
});Observable<String> observable2 = Observable.create(new ObservableOnSubscribe<String>() {
@Override
public void subscribe(ObservableEmitter<String> e) throws Exception {
e.onNext("个");
e.onNext("只");
e.onNext("条");
e.onNext("张");
e.onNext("本");
e.onNext("副");
}
});Observable.zip(observable, observable1, observable2, new Function3<Integer, String, String, String>() {
@Override
public String apply(Integer integer, String s, String s2) throws Exception {
return s + integer + s2;
}
}).subscribe(new Consumer<String>() {
@Override
public void accept(String s) throws Exception {
Log.e("XYK",s);
}
});

Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
for (int i = 0; ; i++) {
Log.e("XYK",i + "");
e.onNext(i);
}
}
});

Observable<Integer> observable = Observable.create(new ObservableOnSubscribe<Integer>() {
@Override
public void subscribe(ObservableEmitter<Integer> e) throws Exception {
for (int i = 0; ; i++) {
e.onNext(i);
}
}
}).filter(new Predicate<Integer>() {
@Override
public boolean test(Integer integer) throws Exception {
return integer % 100 == 0;
}
});

Rxjava的2.x与1.x的区别(官翻)

Nulls

null

null

NullPointerException

Observable.just(null);Single.just(null);Observable.fromCallable(() -> null)
.subscribe(System.out::println, Throwable::printStackTrace);Observable.just(1).map(v -> null)
.subscribe(System.out::println, Throwable::printStackTrace);

Observable<Void>

Observable<Object>

Object

onNext

enum Irrelevant { INSTANCE; }Observable<Object> source = Observable.create((ObservableEmitter<Object> emitter) -> {
System.out.println("Side-effect 1");
emitter.onNext(Irrelevant.INSTANCE);System.out.println("Side-effect 2");
emitter.onNext(Irrelevant.INSTANCE);System.out.println("Side-effect 3");
emitter.onNext(Irrelevant.INSTANCE);
});source.subscribe(e -> { /* Ignored. */ }, Throwable::printStackTrace);

Observable 和 Flowable

Observable

MissingBackpressureException

io.reactivex.Observable

io.reactivex.Flowable

io.reactivex.Observable

Single

Single

onSuccess

onError

SingleObserver

interface SingleObserver<T> {
void onSubscribe(Disposable d);
void onSuccess(T value);
void onError(Throwable error);
}

onSubscribe (onSuccess | onError)?

Completable

Completable

rx.Completable.CompletableSubscriber

io.reactivex.CompletableObserver

onSubscribe(Disposable)

interface CompletableObserver<T> {
void onSubscribe(Disposable d);
void onComplete();
void onError(Throwable error);
}

onSubscribe (onComplete | onError)?

Maybe

Maybe

Single

Completable

Maybe

MaybeSource

MaybeObserver<T>

onSubscribe
(onSuccess | onError | onComplete)?

Maybe

onSubscribe(Disposable)

onXXX

Flowable

onSuccess

onComplete

Flowable

Maybe.just(1)
.map(v -> v + 1)
.filter(v -> v == 1)
.defaultIfEmpty(2)
.test()
.assertResult(2);

Base reactive interfaces

基础reactive接口

Flowable

Publisher<T>

interface ObservableSource<T> {
void subscribe(Observer<? super T> observer);
}interface SingleSource<T> {
void subscribe(SingleObserver<? super T> observer);
}interface CompletableSource {
void subscribe(CompletableObserver observer);
}interface MaybeSource<T> {
void subscribe(MaybeObserver<? super T> observer);
}

Publisher

XSource

Flowable<R> flatMap(Function<? super T, ? extends Publisher<? extends R>> mapper);Observable<R> flatMap(Function<? super T, ? extends ObservableSource<? extends R>> mapper);

Publisher

Flowable

Flowable<Flowable<Integer>> windows = source.window(5);source.compose((Flowable<T> flowable) ->
flowable
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread()));

Subjects 和 Processors

Subject

Observable

Flowable

FlowableProcessor<T>

Subject

T
-> R

io.reactivex.subjects.AsyncSubject

io.reactivex.subjects.BehaviorSubject

io.reactivex.subjects.PublishSubject

io.reactivex.subjects.ReplaySubject

io.reactivex.subjects.UnicastSubject

io.reactivex.processors.AsyncProcessor

io.reactivex.processors.BehaviorProcessor

io.reactivex.processors.PublishProcessor

io.reactivex.processors.ReplayProcessor

io.reactivex.processors.UnicastProcessor

BehaviorProcessor

PublishProcessor

MissingBackpressureException

XProcessor

其他类

rx.observables.ConnectableObservable

io.reactivex.observables.ConnectableObservable<T>

io.reactivex.flowables.ConnectableFlowable<T>

GroupedObservable

rx.observables.GroupedObservable

io.reactivex.observables.GroupedObservable<T>

io.reactivex.flowables.GroupedFlowable<T>

GroupedObservable.from()

GroupedObservable

subscribeActual

class MyGroup<K, V> extends GroupedObservable<K, V> {
final K key;final Subject<V> subject;public MyGroup(K key) {
this.key = key;
this.subject = PublishSubject.create();
}@Override
public T getKey() {
return key;
}@Override
protected void subscribeActual(Observer<? super T> observer) {
subject.subscribe(observer);
}
}

功能接口

Java.util.function.Function

throws Exception

try-catch

Flowable.just("file.txt")
.map(name -> Files.readLines(name))
.subscribe(lines -> System.out.println(lines.size()), Throwable::printStackTrace);

IOException

Files.readLines(name)

Actions

Action3

Action9

ActionN

Action0

io.reactivex.functions.Action

Scheduler

Action1

Consumer

Action2

BiConsumer

ActionN

Consumer<Object[]>

Functions

io.reactivex.functions.Function

io.reactivex.functions.BiFunction

Func3

Func9

Function3

Function9

FuncN

Function<Object[],
R>

Func1<T, Boolean>

Predicate<T>

io.reactivex.functions.Functions

Function<Object[],
R>

Subscriber

org.reactivestreams.Subscription

rx.Producer

rx.Subscription

rx.Subscriber

Flowable.range(1, 10).subscribe(new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
s.request(Long.MAX_VALUE);
}@Override
public void onNext(Integer t) {
System.out.println(t);
}@Override
public void onError(Throwable t) {
t.printStackTrace();
}@Override
public void onComplete() {
System.out.println("Done");
}
});

rx

org.reactivestreams

org.reactivestreams.Subscriber

Flowable

DefaultSubscriber

ResourceSubscriber

DisposableSubscriber

rx.Subscriber

dispose()

ResourceSubscriber<Integer> subscriber = new ResourceSubscriber<Integer>() {
@Override
public void onStart() {
request(Long.MAX_VALUE);
}@Override
public void onNext(Integer t) {
System.out.println(t);
&n
33332
bsp;}@Override
public void onError(Throwable t) {
t.printStackTrace();
}@Override
public void onComplete() {
System.out.println("Done");
}
};Flowable.range(1, 10).delay(1, TimeUnit.SECONDS).subscribe(subscriber);subscriber.dispose();

onCompleted

onComplete

Observable.subscribe(Subscriber)

Subscription

Subscription

CompositeSubscription

CompositeSubscription composite = new CompositeSubscription();composite.add(Observable.range(1, 5).subscribe(new TestSubscriber<Integer>()));由于Reactive-Streams规范，Publisher.subscribe

E subscribeWith(E subscriber)

ResourceSubscriber

Disposable

CompositeDisposable composite2 = new CompositeDisposable();composite2.add(Flowable.range(1, 5).subscribeWith(subscriber));

在onSubscribe/onStart中调用request

Subscriber.onSubscribe

ResourceSubscriber.onStart

request(n)

onNext

Flowable.range(1, 3).subscribe(new Subscriber<Integer>() {@Override
public void onSubscribe(Subscription s) {
System.out.println("OnSubscribe start");
s.request(Long.MAX_VALUE);
System.out.println("OnSubscribe end");
}@Override
public void onNext(Integer v) {
System.out.println(v);
}@Override
public void onError(Throwable e) {
e.printStackTrace();
}@Override
public void onComplete() {
System.out.println("Done");
}
});

OnSubscribe start
1
2
3
Done
OnSubscribe end

onSubscribe

onStart

request

onNext

request

request

Producer

Subscription

Subscription

rx.Subscription

org.reactivestreams.Subscription

rx.Subscription

io.reactivex.Disposable

org.reactivestreams.Publisher

subscribe()

Flowable.subscribe(Subscriber)

Subscription

subscribe

Disposable

Subscription

CompositeSubscription

CompositeDisposable

SerialSubscription

MultipleAssignmentSubscription

SerialDisposable

set()

replace()

RefCountSubscription

背压

Flowable

MissingBackpressureException

onNext

Observable

Reactive-Streams compliance

IllegalArgumentException

RxJavaPlugins.onError

IllegalArgumentException

Subscriber.onError

onSubscribe

onNext

onSubscribe

request(1)

onNext

onNext

request(1)

observeOn

onNext

onSubscribe

onNext

onSubscribe

onSubscribe

Flowable

Runtime hooks

RxJavaPlugins

RxJavaObservableHook

RxJavaHooks

RxJavaPlugins

Schedulers

computation

io

newThread

trampoline

io.reactivex.schedulers.Schedulers

immediate

Scheduler

Schedulers.trampoline()

Schedulers.test()

test()

TestScheduler

new
TestScheduler()

io.reactivex.Scheduler

Worker

public abstract class Scheduler {public Disposable scheduleDirect(Runnable task) { ... }public Disposable scheduleDirect(Runnable task, long delay, TimeUnit unit) { ... }public Disposable scheduleDirectPeriodically(Runnable task, long initialDelay,
long period, TimeUnit unit) { ... }public long now(TimeUnit unit) { ... }// ... rest is the same: lifecycle methods, worker creation
}

Worker

createWorker

now()

TimeUnit

进入reactive的世界

rx.Observable.create()

Flowable

Observable

Single

Maybe

Completable

create

Flowable.create((FlowableEmitter<Integer> emitter) -> {
emitter.onNext(1);
emitter.onNext(2);
emitter.onComplete();
}, BackpressureStrategy.BUFFER);

fromEmitter

Flowable.create

create

离开reactive的世界

Subscriber

Observer

SingleObserver

MaybeObserver

CompletableObserver

subscribe(Consumer<T>, Consumer<Throwable>, Action)

BlockingObservable

blockingX

List<Integer> list = Flowable.range(1, 100).toList().blockingGet(); // toList() returns SingleInteger i = Flowable.range(100, 100).blockingLast();

rx.Subscriber

org.reactivestreams.Subscriber

Subscriber

Observer

Subscriber<Integer> subscriber = new Subscriber<Integer>() {
@Override
public void onSubscribe(Subscription s) {
s.request(Long.MAX_VALUE);
}public void onNext(Integer t) {
if (t == 1) {
throw new IllegalArgumentException();
}
}public void onError(Throwable e) {
if (e instanceof IllegalArgumentException) {
throw new UnsupportedOperationException();
}
}public void onComplete() {
throw new NoSuchElementException();
}
};Flowable.just(1).subscribe(subscriber);

Observer

SingleObserver

MaybeObserver

CompletableObserver

safeSubscribe

subscribe(Consumer<T>, Consumer<Throwable>, Action)

Flowable.just(1)
.subscribe(
subscriber::onNext,
subscriber::onError,
subscriber::onComplete,
subscriber::onSubscribe
);

Testing

Flowable

io.reactivex.subscribers.TestSubscriber

Observable

Single

Maybe

Completable

io.reactivex.observers.TestObserver

test() “operator”

test()

TestSubscriber

TestObserver

TestSubscriber<Integer> ts = Flowable.range(1, 5).test();TestObserver<Integer> to = Observable.range(1, 5).test();TestObserver<Integer> tso = Single.just(1).test();TestObserver<Integer> tmo = Maybe.just(1).test();TestObserver<Integer> tco = Completable.complete().test();

TestSubscriber

TestObserver

assertX

TestSubscriber

TestObserver

Flowable.range(1, 5)
.test()
.assertResult(1, 2, 3, 4, 5)
;

值得注意的新的断言方法

assertResult(T... items)

onComplete

assertFailure(Class<? extends Throwable> clazz, T... items)

assertFailureAndMessage(Class<? extends Throwable> clazz, String message, T... items)

assertFailure

getMessage()

awaitDone(long time, TimeUnit unit)

assertOf(Consumer<TestSubscriber<T>> consumer)

Flowable

Observable

TestSubscriber

TestObserver

提前取消和请求

TestObserver

test()

test(boolean
cancel)

TestSubscriber

TestObserver

PublishSubject<Integer> pp = PublishSubject.create();// nobody subscribed yet
assertFalse(pp.hasSubscribers());pp.test(true);// nobody remained subscribed
assertFalse(pp.hasSubscribers());

TestSubscriber

test(long
initialRequest)

test(long initialRequest, boolean cancel)

TestSubscriber

initialRequest

TestSubscriber

request()

PublishProcessor<Integer> pp = PublishProcessor.create();TestSubscriber<Integer> ts = pp.test(0L);ts.request(1);pp.onNext(1);
pp.onNext(2);ts.assertFailure(MissingBackpressureException.class, 1);

测试异步代码

Flowable.just(1)
.subscribeOn(Schedulers.single())
.test()
.awaitDone(5, TimeUnit.SECONDS)
.assertResult(1);

Mockito & TestSubscriber

Observer

Subscriber.onSubscribe

@SuppressWarnings("unchecked")
public static <T> Subscriber<T> mockSubscriber() {
Subscriber<T> w = mock(Subscriber.class);Mockito.doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock a) throws Throwable {
Subscription s = a.getArgumentAt(0, Subscription.class);
s.request(Long.MAX_VALUE);
return null;
}
}).when(w).onSubscribe((Subscription)any());return w;}

操作符的差别

fromArray

fromIterable

@Beta

@Experimental

1.x Observable 到 2.x Flowable

工厂方法:

amb

amb(ObservableSource...)

bufferSize()

combineLatest

bufferSize

combineLatest(List)

concat

prefetch

concatArray

concatArray

concatArrayDelayError

concatArrayEager

concatArrayEagerDelayError

concatDelayError

concatEagerDelayError

create(SyncOnSubscribe)

generate

create(AsnycOnSubscribe)

create(OnSubscribe)

create(FlowableOnSubscribe, BackpressureStrategy)

unsafeCreate()

from

fromArray

fromIterable

fromFuture

fromPublisher

fromAsync

create()

intervalRange()

limit

take

merge

prefetch

mergeDelayError

prefetch

sequenceEqual

bufferSize

switchOnNext

prefetch

switchOnNextDelayError

prefetch

timer

zip

bufferSize

delayErrors

zipArray

zipIterable

实例方法:

all

Single<Boolean>

any

Single<Boolean>

asObservable

hide()

buffer

Collection

cache(int)

collect

Single<U>

collect(U, Action2<U, T>)

collectInto

Single<U>

concatMap

prefetch

concatMapDelayError

prefetch

concatMapEager

prefetch

concatMapEagerDelayError

prefetch

count

Single<Long>

countLong

count

distinct

Collection

doOnCompleted

doOnComplete

doOnUnsubscribe

Flowable.doOnCancel

doOnDispose

doOnLifecylce

onSubscribe

request

cancel

elementAt(int)

elementAt(Func1, int)

filter(predicate).elementAt(int)

elementAtOrDefault(int, T)

elementAt(int, T)

Single<T>

elementAtOrDefault(Func1, int, T)

filter(predicate).elementAt(int, T)

first()

firstElement

Maybe<T>

first(Func1)

filter(predicate).first()

firstOrDefault(T)

first(T)

Single<T>

firstOrDefault(Func1, T)

filter(predicate).first(T)

flatMap

prefetch

forEachWhile(Predicate<T>, [Consumer<Throwable>, [Action]])

groupBy

bufferSize

delayError

ignoreElements

Completable

isEmpty

Single<Boolean>

last()

lastElement

Maybe<T>

last(Func1)

filter(predicate).last()

lastOrDefault(T)

last(T)

Single<T>

lastOrDefault(Func1, T)

filter(predicate).last(T)

nest

just

publish(Func1)

prefetch

reduce(Func2)

Maybe<T>

reduceWith(Callable, BiFunction)

Single<T>

repeatUntil(BooleanSupplier)

repeatWhen(Func1, Scheduler)

subscribeOn(Scheduler).repeatWhen(Function)

retry

retry(Predicate)

retry(int,
Predicate)

retryUntil(BooleanSupplier)

retryWhen(Func1, Scheduler)

subscribeOn(Scheduler).retryWhen(Function)

sampleWith(Callable, BiFunction)

single()

singleElement

Maybe<T>

single(Func1)

filter(predicate).single()

singleOrDefault(T)

single(T)

Single<T>

singleOrDefault(Func1, T)

filter(predicate).single(T)

skipLast

bufferSize

delayError

startWith

startWithArray

startWithArray

subscribeWith

switchMap

prefetch

switchMapDelayError

prefetch

takeLastBuffer

test()

timeout(Func0<Observable>, ...)

timeout(Publisher, ...)

defer(Callable<Publisher>>)

toBlocking().y

blockingY()

toFuture

toCompletable

ignoreElements

toList

Single<List<T>>

toMap

Single<Map<K, V>>

toMultimap

Single<Map<K, Collection<V>>>

toFuture

toObservable

toSingle

single(T)

toSortedList

Single<List<T>>

withLatestFrom

zipWith

prefetch

delayErrors

不同的返回类型

Single

all(Predicate)

Observable<Boolean>

Single<Boolean>

any(Predicate)

Observable<Boolean>

Single<Boolean>

count()

Observable<Long>

Single<Long>

elementAt(int)

Observable<T>

Maybe<T>

elementAt(int, T)

Observable<T>

Single<T>

first(T)

Observable<T>

Single<T>

NoSuchElementException

firstElement()

Observable<T>

Maybe<T>

ignoreElements()

Observable<T>

Completable

isEmpty()

Observable<Boolean>

Single<Boolean>

last(T)

Observable<T>

Single<T>

lastElement()

Observable<T>

Maybe<T>

reduce(BiFunction)

Observable<T>

Maybe<T>

reduce(Callable, BiFunction)

Observable<U>

Single<U>

reduceWith(U, BiFunction)

Observable<U>

Single<U>

single(T)

Observable<T>

Single<T>

singleElement()

Observable<T>

Maybe<T>

toList()

Observable<List<T>>

Single<List<T>>

List

toMap()

Observable<Map<K, V>>

Single<Map<K, V>>

Map

toMultimap()

Observable<Map<K, Collection<V>>>

Single<Map<K, Collection<V>>>

Map

toSortedList()

Observable<List<T>>

Single<List<T>>

List

移除

Flowable.toCompletable()

Flowable.ignoreElements()

Flowable.toSingle()

Flowable.single(T)

Flowable.toMaybe()

Flowable.singleElement()

Observable.toCompletable()

Observable.ignoreElements()

Observable.toSingle()

Observable.single(T)

Observable.toMaybe()

Observable.singleElement()

其他改变

doOnCancel/doOnDispose/unsubscribeOn

doOnUnsubscribe

SafeSubscriber

unsubscribe

Subscriber

Subscription

cancel()

unsubscribeOn

cancel

unsubscribeOn

doOnCancel

Flowable.just(1, 2, 3)
.doOnCancel(() -> System.out.println("Cancelled!"))
.subscribe(System.out::println);

take

onNext

Flowable.just(1, 2, 3)
.doOnCancel(() -> System.out.println("Cancelled!"))
.take(2)
.subscribe(System.out::println);

using