Observable.just(new User("白瑞德"))
                  .map(new Function<User, String>() {
                      @Override
                      public String apply(User user) throws Throwable {
                          return user.getName();
                      }
                  })
                  .subscribe(new Consumer<String>() {
                      @Override
                      public void accept(String s) throws Throwable {
                          System.out.println(s);
                      }
                  });
<<<白瑞德

class User {
    private String       name;
    private List<String> book;
}

Observable.fromIterable(userList)
                  .map(new Function<User, List<String>>() {
                      @Override
                      public List<String> apply(User user) throws Throwable {
                          return user.getBook();
                      }
                  })
                  .subscribe(new Consumer<List<String>>() {
                      @Override
                      public void accept(List<String> strings) throws Throwable {
                          for (String s : strings) {
                              System.out.println(s);
                          }
                      }
                  });

Observable.fromIterable(userList)
                  .flatMap(new Function<User, ObservableSource<String>>() {
                      @Override
                      public ObservableSource<String> apply(User user) throws Throwable {
                          return Observable.fromIterable(user.getBook());
                      }
                  })
                  .subscribe(new Consumer<String>() {
                      @Override
                      public void accept(String  o) throws Throwable {
                          System.out.println(o);
                      }
                  });

public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
    //接受一個(gè)Function實(shí)例，并返回一個(gè)ObservableMap
    return new ObservableMap<T, R>(this, mapper);
}
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) {
        //調(diào)用用父類構(gòu)造方法，初始化父類中的downstream
        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) {
            v = mapper.apply(t);
            downstream.onNext(v);
        }
    }
}

public final <R> Observable<R> flatMap(Function<? super T, ? extends ObservableSource<? extends R>> mapper) {
        return new ObservableFlatMap<T, R>(this, mapper, delayErrors, maxConcurrency, bufferSize);
}
public final class ObservableFlatMap<T, U> extends AbstractObservableWithUpstream<T, U> {
    final Function<? super T, ? extends ObservableSource<? extends U>> mapper;
    final boolean delayErrors;
    final int maxConcurrency;
    final int bufferSize;
    public ObservableFlatMap(ObservableSource<T> source,
            Function<? super T, ? extends ObservableSource<? extends U>> mapper,
            boolean delayErrors, int maxConcurrency, int bufferSize) {
        super(source);
    }
    @Override
    public void subscribeActual(Observer<? super U> t) {
        source.subscribe(new MergeObserver<T, U>(t, mapper, delayErrors, maxConcurrency, bufferSize));
    }
    static final class MergeObserver<T, U> extends AtomicInteger implements Disposable, Observer<T> {
        MergeObserver(Observer<? super U> actual, Function<? super T, ? extends ObservableSource<? extends U>> mapper,
                boolean delayErrors, int maxConcurrency, int bufferSize) {
            ...   
            this.observers = new AtomicReference<InnerObserver<?, ?>[]>(EMPTY);
        }
        @Override
        public void onSubscribe(Disposable d) {
            downstream.onSubscribe(this);
        }
        @Override
        public void onNext(T t) {
            ObservableSource<? extends U> p;
            p = mapper.apply(t);    
            subscribeInner(p);
        }
        @SuppressWarnings("unchecked")
        void subscribeInner(ObservableSource<? extends U> p) {
                InnerObserver<T, U> inner = new InnerObserver<T, U>(this, uniqueId++);
                p.subscribe(inner);
        }
        void drain() {    
            drainLoop();
        }
        void drainLoop() {
            final Observer<? super U> child = this.downstream;
            child.onNext(o);
        }
    }
    static final class InnerObserver<T, U> extends AtomicReference<Disposable>
    implements Observer<U> {
        private static final long serialVersionUID = -4606175640614850599L;
        final long id;
        final MergeObserver<T, U> parent;
        volatile boolean done;
        volatile SimpleQueue<U> queue;
        int fusionMode;
        InnerObserver(MergeObserver<T, U> parent, long id) {
            this.id = id;
            this.parent = parent;
        }
        @Override
        public void onNext(U t) {
            parent.drain();
        }
    }
}