Android：rxjava简单实现原理（map/flatmap操作符）

Posted 2021-09-03 JMW1407

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Android：rxjava简单实现原理（map/flatmap操作符）相关的知识，希望对你有一定的参考价值。

rxjava

装饰者模式
rxjava装饰者模式
- 1、rxjava中转换操作符map的简单实现
- 2、rxjava中转换操作符flatmap的简单实现

装饰者模式

1、背景

假设奶茶店有两种茶，果茶（fruit tea）和奶茶(milky tea)，同时这两种茶饮料可以添加不同的配料果肉（pulp）或者蔗糖（sugar），茶品可以和配料进行组合，所以可以得到：

1、pulp_fruit_tea（果茶加果肉）
2、pulp_milky_tea（奶茶加果肉）
3、sugar_fruit_tea（果茶加糖）
4、sugar_milky_tea（奶茶加糖）
5、 fruit_tea（果茶）
6、milky_tea（奶茶）
日后还会增加新的茶品和配料。

1、采用单一继承的方式的UML图： 有多少个技能就写多少个子类来继承这个tea类。

可以明显发现一个问题：就是随着物品修饰种类的增加，继承的类也越来越多，每增加一个技能可以组合出N个子类，那不就要加到天荒地老？
2、直接抽象一个tea类，里面包含了所有的配料。

这样子有多少个角色就增加多少个子类就可以了，不用根据技能增加类，避免造成子类爆炸，但是，每个角色的技能是可以叠加使用的，角色越多或者技能叠加种类越多，那么就要在超类增加越多的方法，而且直接修改超类不符合开闭原则（超类对扩展开放，对修改关闭）。

综上：

装饰模式实际上是一直提倡的组合代替继承的实践方式,个人认为要理解装饰者模式首先需要理解为什么需要组合代替继承,继承又是为什么让人深恶痛绝.

为什么建议使用组合代替继承?
面向对象的特性有继承与封装,但两者却又有一点矛盾,继承意味子类依赖了父类中的实现,一旦父类中改变实现则会对子类造成影响,这是打破了封装性的一种表现. 而组合就是巧用封装性来实现继承功能的代码复用.

2、定义

1.定义：

-装饰器模式又名包装(Wrapper)模式。动态地给一个对象添加一些额外的职责。就增加功能来说，Decorator模式相比生成子类更为灵活。

装饰器模式以对客户端透明的方式拓展对象的功能，是继承关系的一种替代方案。

UML关系图说明：

继承关系：实线、空三角箭头，具体装饰继承Decorator，Decorator继承Component；聚合：实线、菱形

1、抽象构件(Component)角色：一个抽象接口。装饰对象和被装饰对象【具体组件对象】共有的父类接口。这样客户端对象就能以相同的方式操作具体组件对象和装饰对象【或者说可以将装饰类作为组件类看待】
2、具体构件(ConcreteComponent)角色：定义一个将要接收附加责任的类。装饰模式是为这个基类动态添加新功能。
3、装饰(Decorator)角色：装饰对象包含一个真实组件对像的引用。它的作用是接受需要扩展功能的具体组件类；实现抽象组件接口，使得在动态添加功能时具体装饰类和具体组件类用法相同，使模式更加灵活。
4、具体装饰(ConcreteDecorator)角色：负责给构件对象“贴上”附加的责任。

3、特征

需要扩展一个类的功能或给一个类增加附加责任。
需要动态地给一个对象增加功能，这些功能可以再动态地撤销。
需要增加由一些基本功能的排列组合而产生的非常大量的功能

优点：

1、装饰这模式和继承的目的都是扩展对象的功能，但装饰者模式比继承更灵活通过使用不同的具体装饰类以及这些类的排列组合，设计师可以创造出很多不同行为的组合装饰者模式有很好地可扩展性
2、装饰类和被装饰类可以独立发展，而不会相互耦合。Component类无须知道Decorator类，而Decorator也不用知道具体的构件，用户可以根据需要增加新的具体构件类和具体装饰类，在使用时再对其进行组合，原有代码无须改变，符合“开闭原则”。

缺点：

装饰者模式会导致设计中出现许多小对象，如果过度使用，会让程序变的更复杂。并且更多的对象会是的差错变得困难，特别是这些对象看上去都很像。

4、装饰者模式demo

//被装饰者的对象接口
public interface Drink {
    public float cost();
    public String desc();
}

//具体的被装饰者
public class Dounai implements Drink{
    @Override
    public float cost() {
        return 3f;
    }
    @Override
    public String  desc() {
        return "纯豆奶";

    }
}

//装饰者的基类
public abstract class Decroator implements Drink  {
    private Drink drink; //要装饰的对象

    public Decroator(Drink drink) {
        this.drink = drink;
    }

    @Override
    public float cost() {
        return drink.cost();
    }

    @Override
    public String desc() {
      return  drink.desc();
    }
}

//具体的装饰者
public class Blackdou extends Decroator {
    public Blackdou(Drink drink) {
        super(drink);
    }

    @Override
    public float cost() {
        return super.cost()+2f;
    }

    @Override
    public String desc() {
        return super.desc()+"+黑豆";
    }
}

//具体的装饰者类
public class SugarDecroator extends Decroator {
    public SugarDecroator(Drink drink) {
        super(drink);
    }
    @Override
    public float cost() {
        return super.cost()+1f;
    }
    @Override
    public String desc() {
        return super.desc()+"+糖";
    }
}

//具体的装饰者类
public class SugarDecroator extends Decroator {
    public SugarDecroator(Drink drink) {
        super(drink);
    }
    @Override
    public float cost() {
        return super.cost()+1f;
    }
    @Override
    public String desc() {
        return super.desc()+"+糖";
    }
}

实验结果

这杯豆浆价格为：6.0
描述为：纯豆奶+黑豆+糖

rxjava装饰者模式

1、rxjava中转换操作符map的简单实现

Observable类

/**
 * 被观察者的核心抽象类
 * 也是使用框架的入口
 * @param <T>
 */
public abstract class Observable<T> implements ObservableSource<T> {
    @Override
    public void subscribe(Observer observer) {
        // 和谁建立订阅？
        // 怎么建立订阅？
        // 为了保证拓展性，交给具体的开发人员实现。这里提供一个抽象的方法
        subscribeActual(observer);
    }
    protected abstract void subscribeActual(Observer<T> observer);
	//creat操作符
    public  static <T> Observable<T> create(ObservableOnSubscribe<T> source){
        return new ObservableCreate<>(source);
    }
    //map操作符
	public <R> ObservableMap<T, R> map(Function<T, R> function) {
        return new ObservableMap<>(this, function);
    }
    //flatmap操作符
    public <R> ObservableFlatMap<T, R> flatMap(Function<T, ObservableSource<R>> function) {
        return new ObservableFlatMap<>(this, function);
    }
}

Observer接口

public interface Observer<T> {
    void onSubscribe();//建立订阅关系时候的回掉方法，什么时候和被观察者建立订阅关系，就会调用该方法
    void onNext(T t);
    void onComplete();
    void onError(Throwable throwable);

}

ObservableCreate类

/**
 * 1、创建一个被观察者
 * 2、被观察者发射事件由具体的either发射器来完成
 * @param <T>
 */
public class ObservableCreate<T> extends Observable<T> {
    final ObservableOnSubscribe<T> source;

    public ObservableCreate(ObservableOnSubscribe<T> source) {
        this.source = source;//创建一个被观察者
    }

    @Override
    protected void subscribeActual(Observer<T> observer) {
        observer.onSubscribe();//建立订阅的时候调用
        CreateEmitter<T> emitter = new CreateEmitter<T>(observer);
        source.subscribe(emitter);
    }

    static class CreateEmitter<T> implements Emitter<T>{
        Observer<T>observer;//这里持有一个观察者，当事件发生时，直接调用该观察者对事件进行调用即可。
        boolean done;//互斥实现

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

        public void onNext(T t){
            if(done) return;
            observer.onNext(t);
        }

        public void onError(Throwable throwable){
            if(done) return;
            observer.onError(throwable);
            done = true;
        }

        public void onComplete(){
            if(done) return;
            observer.onComplete();
            done = true;
        }
    }

}

Emitter接口

/**
 * 事件发射器
 * @param <T>
 */
public interface Emitter<T> {
    void onNext(T t);
    void onComplete();
    void onError(Throwable throwable);
}

ObservableOnSubscribe接口

/**
 * 被观察者和事件发射器建立关系
 * 被观察者和事件之间解偶
 * @param <T>
 */
public interface ObservableOnSubscribe<T> {
    void subscribe(Emitter<T> emitter);
}

ObservableSource接口

/**
 * 被观察者的顶层接口
 * @param <T>
 */

public interface ObservableSource<T> {
    void subscribe(Observer<T>observer);
}

Function接口

/**
 * 数据源转换函数
 * @param <T>
 * @param <R>
 */
public interface Function<T,R>{

    R apply(T t);
}

ObservableMap

public class ObservableMap<T, U> extends AbstractObservableWithUpStream<T, U> {
    Function<T, U> function;

    public ObservableMap(ObservableSource<T> source, Function<T, U> function) {
        super(source);
        this.function = function;
    }

    @Override
    protected void subscribeActual(Observer<U> observer) {
        source.subscribe(new MapObserver<>(observer,function));
    }

    static class MapObserver<T, U> implements Observer<T> {
        final Observer<U> downStream;
        final Function<T, U> mapper;

        public MapObserver(Observer<U> downStream, Function<T, U> mapper) {
            this.downStream = downStream;
            this.mapper = mapper;
        }

        @Override
        public void onSubscribe() {
            downStream.onSubscribe();
        }

        @Override
        public void onNext(T t){
        //map操作符的具体实现
            U u = mapper.apply(t);
            downStream.onNext(u);
        }

        @Override
        public void onComplete(){
            downStream.onComplete();
        }

        @Override
        public void onError(Throwable throwable){
            downStream.onError(throwable);
        }
    }

}

AbstractObservableWithUpStream

/**
 * 抽象装饰类
 * @param <T>
 * @param <U>
 */
public abstract class AbstractObservableWithUpStream<T,U> extends Observable<U>{
    protected final ObservableSource<T> source;//在这个基础上进行装饰

    public AbstractObservableWithUpStream(ObservableSource<T>source){
        this.source = source;
    }

}

RxjavaTest

public class RxjavaTest {
    public static void main(String[] args){
        Observable.create(new ObservableOnSubscribe<Object>() {
            @Override
            public void subscribe(Emitter<Object> emitter){
                System.out.println("subscribe:.....");
                emitter.onNext("aaaa");
                emitter.onNext("CCCC");
                emitter.onNext("ddddd");
                emitter.onError(new Throwable());
                emitter.onComplete();
            }

        }).subscribe(new Observer() {
            @Override
            public void onSubscribe() {
                System.out.println("onSubscribe...");
            }

            @Override
            public void onNext(Object o) {
                System.out.println("onNext:.... " + o);
            }

            @Override
            public void onComplete() {
                System.out.println("onComplete:... " );
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("onError :... " );
            }
        });
    }

}

实验结果

onSubscribe...
subscribe:.....
onNext:.... 处理后的：aaaa
onNext:.... 处理后的：CCCC
onNext:.... 处理后的：ddddd
onError :...

2、rxjava中转换操作符flatmap的简单实现

ObservableFlatMap

public class ObservableFlatMap<T, U> extends AbstractObservableWithUpStream<T, U> {

    Function<T, ObservableSource<U>> function;

    public ObservableFlatMap(ObservableSource<T> source, Function<T, ObservableSource<U>> function) {
        super(source);
        this.function = function;
    }

    @Override
    protected void subscribeActual(Observer<U> observer) {
        source.subscribe(new MergeObserver<>(observer, function));
    }

    static class MergeObserver<T, U> implements Observer<T> {

        final Observer<U> downStream;

        final Function<T, ObservableSource<U>> mapper;

        public MergeObserver(Observer<U> downStream, Function<T, ObservableSource<U>> mapper) {
            this.downStream = downStream;
            this.mapper = mapper;
        }

        @Override
        public void onSubscribe() {
            downStream.onSubscribe();
        }

        @Override
        public void onNext(T t) {
            ObservableSource<U> observable = mapper.apply(t);
            observable.subscribe(new Observer<U>() {
                @Override
                public void onSubscribe() {

                }

                @Override
                public void onNext(U u) {
                    downStream.onNext(u);
                }

                @Override
                public void onComplete() {

                }

                @Override
                public void onError(Throwable throwable) {

                }
            });
        }

        @Override
        public void onComplete() {
            downStream.onComplete();
        }

        @Override
        public void onError(Throwable throwable) {
            downStream.onError(throwable);
        }
    }


}

RxjavaTest

public class RxjavaTest {
    public static void main(String[] args) {
        Observable.create(new ObservableOnSubscribe<Object>() {
                    @Override
                    public void subscribe(Emitter<Object> emitter) {
                        System.out.println("subscribe:.....");
                        emitter.onNext("aaaa");
                        emitter.onNext("CCCC");
                        emitter.onNext("ddddd");
                        emitter.onError(new Throwable());
                        emitter.onComplete();
                    }

                }).flatMap(new Function<Object, ObservableSource<Object>>() {
                    @Override
                    public ObservableSource<Object> apply(Object o) {
                        return Observable.create(new ObservableOnSubscribe<Object>() {
                            @Override
                            public void subscribe(Emitter<Object> emitter) {
                                emitter.onNext("处理后的" + o);
                            }
                        });
                    }
                })
                .subscribe(new Observer() {
                    @Override
                    public void onSubscribe() {
                        System.out.println("onSubscribe...");
                    }

                    @Override
                    public void onNext(Object o) {
                        System.out.println("onNext:.... " + o);
                    }

                    @Override
                    public void onComplete() {
                        System.out.println("onComplete:... ");
                    }

                    @Override
                    public void onError(Throwable throwable) {
                        System.out.println("onError :... ");
                    }
                });
    }

}

实验结果

在onSubscribe...
subscribe:.....
onNext:.... 处理后的aaaa
onNext:.... 处理后的CCCC
onNext:.... 处理后的ddddd
onError :... 

Process finished with exit code 0

以上是关于Android：rxjava简单实现原理（map/flatmap操作符）的主要内容，如果未能解决你的问题，请参考以下文章

Rxjava源码分析&实践实践环节：map操作符功能实现

Android异步框架RxJava 1.x系列 - 事件及事件序列转换原理

Android RxJava入门教程 & 简单原理分析

Android OkHttp+Retrofit+Rxjava+Hilt实现网络请求框架