对Observable发射的 每一项数据应用一个函 数,执行变换操作,如下图。

map操作符对原始Observable发射的的 每一项数据应用一个你选择的函数 ,然后返回一个发射这些结果的Observable。
RxJava将 这个操作符实现为map函数 ,这个操作符默认 不再任何特定的调度器 上执行。
(二)示例
public void mapTest(){ Observable.just("HELLO") .map(new Function<String, String>() { @Override public String apply(String s) throws Exception { return s.toLowerCase(); } }) .map(new Function<String, String>() { @Override public String apply(String s) throws Exception { return s + " world"; } }) .subscribe(new Consumer<String>() { @Override public void accept(String s) throws Exception { System.out.println(s); } }); } 复制代码
输出结果:
hello world 复制代码
(三)源码分析
1.调用map()方法
/** * Returns an Observable that applies a specified function to each item emitted by the source ObservableSource and * emits the results of these function applications. * * @param <R> the output type * @param mapper * a function to apply to each item emitted by the ObservableSource * @return an Observable that emits the items from the source ObservableSource, transformed by the specified * function */ @CheckReturnValue @SchedulerSupport(SchedulerSupport.NONE) public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) { ObjectHelper.requireNonNull(mapper, "mapper is null"); return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper)); } 复制代码
2.底层在调用ObservableMap类
将自定义的Function对象赋值给成员变量:function. 该类中同样有个重写的subscribeActual()方法, 在
@Override public void subscribeActual(Observer<? super U> t) { source.subscribe(new MapObserver<T, U>(t, function)); } 复制代码
后续调用subscribe()时,其会先调用Observable类中的subscribe()方法
/**
* Subscribes to an ObservableSource and provides a callback to handle the items it emits.
*
* @param onNext
* the {@code Consumer<T>} you have designed to accept emissions from the ObservableSource
* @return a {@link Disposable} reference with which the caller can stop receiving items before
* the ObservableSource has finished sending them
* @throws NullPointerException
* if {@code onNext} is null
*/
@CheckReturnValue
@SchedulerSupport(SchedulerSupport.NONE)
public final Disposable subscribe(Consumer<? super T> onNext) {
return subscribe(onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION, Functions.emptyConsumer());
}
复制代码
subscribe(onNext, Functions.ON_ERROR_MISSING, Functions.EMPTY_ACTION, Functions.emptyConsumer())
该方法会调用Observable同类中的重载的方法:subscribe()
public final Disposable subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError,
Action onComplete, Consumer<? super Disposable> onSubscribe) {
ObjectHelper.requireNonNull(onNext, "onNext is null");
ObjectHelper.requireNonNull(onError, "onError is null");
ObjectHelper.requireNonNull(onComplete, "onComplete is null");
ObjectHelper.requireNonNull(onSubscribe, "onSubscribe is null");
LambdaObserver<T> ls = new LambdaObserver<T>(onNext, onError, onComplete, onSubscribe);
// 核心方法
subscribe(ls);
return ls;
}
复制代码
该方法又会调用同类中重载的subscribe()方法
@SchedulerSupport(SchedulerSupport.NONE)
@Override
public final void subscribe(Observer<? super T> observer) {
ObjectHelper.requireNonNull(observer, "observer is null");
try {
observer = RxJavaPlugins.onSubscribe(this, observer);
ObjectHelper.requireNonNull(observer, "Plugin returned null Observer");
// 核心方法,继承Observable的类都须该方法
subscribeActual(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 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;
}
}
复制代码
此时就会调用 ObservableMap 类中的 subscribeActual() 方法实现,
该方法又会调用上述方法,不过此时subscribeActual()方法是ObservableJust类中的方法
@Override
protected void subscribeActual(Observer<? super T> s) {
ScalarDisposable<T> sd = new ScalarDisposable<T>(s, value);
s.onSubscribe(sd);
sd.run();
}
复制代码
该方法就会使用s.onSubscribe()来BasicFuseableObserver类中的OnSubscribe()方法
// final: fixed protocol steps to support fuseable and non-fuseable upstream
@SuppressWarnings("unchecked")
@Override
public final void onSubscribe(Disposable s) {
if (DisposableHelper.validate(this.s, s)) {
this.s = s;
if (s instanceof QueueDisposable) {
this.qs = (QueueDisposable<T>)s;
}
if (beforeDownstream()) {
//对应的代码
actual.onSubscribe(this);
afterDownstream();
}
}
}
复制代码
进过多次的调用,最终就会调用我们再主方法中重写的accept()方法,输出对应的结果。
二.flatmap操作符
(一)定义
flatMap将一个发射数据的Observable变换为多个Observable,然后将它们发射的数据 合并放进一个单独的Observable ,如图2:

flapMap操作符使用一个指定的函数对原始Observable发射的每一项数据执行变换操作,这个函数返回一个本身也发射数据的Observable,然后flatMap合并这些Observables发射的数据,最后将合并后的结果当作它自己的数据序列发射。
(二)示例
数据类
public class User { public String userName; public List<Address> addresses; public static class Address { public String street; public String city; } } 复制代码
public void flatMapTest() {
User user = new User();
user.userName = "tony";
user.addresses = new ArrayList<>();
User.Address address1 = new User.Address();
address1.street = "ren ming road";
address1.city = "Su zhou";
user.addresses.add(address1);
User.Address address2 = new User.Address();
address2.street = "dong wu bei road";
address2.city = "Su zhou";
user.addresses.add(address2);
Observable.just(user)
.flatMap(new Function<User, ObservableSource<User.Address>>() {
@Override
public ObservableSource<User.Address> apply(User user) throws Exception {
return Observable.fromIterable(user.addresses);
}
})
.subscribe(new Consumer<User.Address>() {
@Override
public void accept(User.Address address) throws Exception {
System.out.println(address.street);
}
});
}
复制代码
输出结果:
ren ming road dong wu bei road 复制代码
(三) 源码分析
1.flatMap()
flatMap()底层会调用有多个重载的方法,最终会调用如下方法:
/** * Returns an Observable that emits items based on applying a function that you supply to each item emitted * by the source ObservableSource, where that function returns an ObservableSource, and then merging those resulting * ObservableSources and emitting the results of this merger, while limiting the maximum number of concurrent * subscriptions to these ObservableSources. * * @param <R> the value type of the inner ObservableSources and the output type * @param mapper * a function that, when applied to an item emitted by the source ObservableSource, returns an * ObservableSource * @param maxConcurrency * the maximum number of ObservableSources that may be subscribed to concurrently * @param delayErrors * if true, exceptions from the current Observable and all inner ObservableSources are delayed until all of them terminate * if false, the first one signalling an exception will terminate the whole sequence immediately * @param bufferSize * the number of elements to prefetch from each inner ObservableSource * @return an Observable that emits the result of applying the transformation function to each item emitted * by the source ObservableSource and merging the results of the ObservableSources obtained from this * transformation */ @CheckReturnValue @SchedulerSupport(SchedulerSupport.NONE) public final <R> Observable<R> flatMap(Function<? super T, ? extends ObservableSource<? extends R>> mapper, boolean delayErrors, int maxConcurrency, int bufferSize) { ObjectHelper.requireNonNull(mapper, "mapper is null"); ObjectHelper.verifyPositive(maxConcurrency, "maxConcurrency"); ObjectHelper.verifyPositive(bufferSize, "bufferSize"); if (this instanceof ScalarCallable) { @SuppressWarnings("unchecked") T v = ((ScalarCallable<T>)this).call(); if (v == null) { return empty(); } return ObservableScalarXMap.scalarXMap(v, mapper); } return RxJavaPlugins.onAssembly(new ObservableFlatMap<T, R>(this, mapper, delayErrors, maxConcurrency, bufferSize)); } 复制代码
而后根据this是否为ScalarCallable类,返回不同的对象:
本示例是该示例,进而调用 ObservableScalarXMap.scalarXMap(v, mapper);
/** * Maps a scalar value into an Observable and emits its values. * * @param <T> the scalar value type * @param <U> the output value type * @param value the scalar value to map * @param mapper the function that gets the scalar value and should return * an ObservableSource that gets streamed * @return the new Observable instance */ public static <T, U> Observable<U> scalarXMap(T value, Function<? super T, ? extends ObservableSource<? extends U>> mapper) { return RxJavaPlugins.onAssembly(new ScalarXMapObservable<T, U>(value, mapper)); } 复制代码
进而创建ScalarXMapObservable对象。
2.subscribe()
该方法也有多个重载方法,最终会调用 subscribe(Consumer<? super T> onNext, Consumer<? super Throwable> onError, Action onComplete, Consumer<? super Disposable> onSubscribe)
并执行其中的**subscribe(ls); 方法 进而再调用 subscribe(Observer<? super T> observer)
并执行其中的 subscribeActual(observer);**方法
进而调用 ObservableScalarXMap
的**subscribeActual(Observer<? super R> s)**方法
@SuppressWarnings("unchecked") @Override public void subscribeActual(Observer<? super R> s) { ObservableSource<? extends R> other; try { // 核心代码 // 这儿调用在用户自定义的flatMap方法中自定义apply()方法 other = ObjectHelper.requireNonNull(mapper.apply(value), "The mapper returned a null ObservableSource"); } catch (Throwable e) { EmptyDisposable.error(e, s); return; } // 此处是判断other的类型,进而执行不同的方法 if (other instanceof Callable) { R u; try { u = ((Callable<R>)other).call(); } catch (Throwable ex) { Exceptions.throwIfFatal(ex); EmptyDisposable.error(ex, s); return; } if (u == null) { EmptyDisposable.complete(s); return; } ScalarDisposable<R> sd = new ScalarDisposable<R>(s, u); s.onSubscribe(sd); sd.run(); } else { // 因为other是Observable类型,因而执行这步 other.subscribe(s); } } } 复制代码
在subscribeActual()调用 mapper.apply(value)
执行 用户自定义的flatMap方法中的apply()方法 。
此时又会调用Observable类的subscribe()方法。
注意此处的subscribe()和上面已经调用的Observable不是同一个对象!!!
再次调用该方法中的subscribeActual()方法,进而调用 ObservableFromIterable
类中的**subscribeActual()**方法:
@Override public void subscribeActual(Observer<? super T> s) { Iterator<? extends T> it; try { // 此处对source的数据进行依次发送 it = source.iterator(); } catch (Throwable e) { Exceptions.throwIfFatal(e); EmptyDisposable.error(e, s); return; } boolean hasNext; try { // 判断是否it ------------------------------------------ hasNext = it.hasNext(); } catch (Throwable e) { Exceptions.throwIfFatal(e); EmptyDisposable.error(e, s); return; } if (!hasNext) { EmptyDisposable.complete(s); return; } FromIterableDisposable<T> d = new FromIterableDisposable<T>(s, it); s.onSubscribe(d); if (!d.fusionMode) { d.run(); } } 复制代码
原文
https://juejin.im/post/5ef376705188252e5a5dacc7
本站部分文章源于互联网,本着传播知识、有益学习和研究的目的进行的转载,为网友免费提供。如有著作权人或出版方提出异议,本站将立即删除。如果您对文章转载有任何疑问请告之我们,以便我们及时纠正。PS:推荐一个微信公众号: askHarries 或者qq群:474807195,里面会分享一些资深架构师录制的视频录像:有Spring,MyBatis,Netty源码分析,高并发、高性能、分布式、微服务架构的原理,JVM性能优化这些成为架构师必备的知识体系。还能领取免费的学习资源,目前受益良多

转载请注明原文出处:Harries Blog™ » RxJava的部分变换操作符源码分析