RxJava编程思想1-（实现简易版Rxjava，如何基本功能和链式调用？）

Posted 2022-12-08 0 and 1

前言

已经使用rxjava两个月了，觉得rxjava特别好用，爱不释手。本文目的是通过几百行的代码，帮助大家理解rxjava中的链式调用，操作符，线程切换是如何实现的。如果有写的不对的地方欢迎在评论区留言，如果觉得写的可以，请点赞，关注，谢谢。

代码链接： github

目录：

RxJava编程思想1-（实现简易版Rxjava，如何基本功能和链式调用？）
RxJava编程思想2-（实现简易版Rxjava，如何实现操作符？）
RxJava编程思想3-（实现简易版Rxjava，如何实现线程切换？）

Rxjava涉及的几个概念

Observable ：被观察者，用来生产发送事件；
Observer：观察者，接收被观察者传来的事件；
Data：数据，被观察者发出数据并传递给观察者；（数据传递过程可以被各种操作符处理，这里为了帮助理解，可以先忽略操作符的存在）
Subscribe：被观察者和观察者通过订阅产生关系后，才具备事件发送和接收能力；

提醒：代码架构一般都少不了接口，要有面向接口编程的思想。接口约定了不同模块如何交互。
Rxjava使用了很多泛型，如果不明白，推荐《java编程思想》这本经典书籍。

如果要实现上面的功能，你会如何编码？

首先我们定义一个观察者：

/**
 * 定义观察者的数据回调接口
 *
 * @param <T>
 */
public interface Observer<T> 
    void onSubscribe();

    void onNext(T value);

    void onError(Throwable e);

    void onComplete();

定义一个ObservableSource 接口，约定被观察者的行为

/**
 * 定义被观察者的行为
 * @param <T>
 */
public interface ObservableSource<T> 
    void subscribe(Observer<? super T> observer);

然后我们约定一个发射数据的对象的行为（

//定义发射数据的行为
public interface Emitter<T> 
    void onNext(T value);

    void onError(Throwable error);

    void onComplete();

定义被被观察者发射数据的特定行为（这里说一下，rxjava源码里面有一个serialize()方法，这里先去掉，目的是方便大家理解）

/**
 * 定义被观察者特定的事件发射行为
 *
 * @param <T>
 */
public interface ObservableEmitter<T> extends Emitter<T> 

定义被观察者Observable，提供发射数据的方法：通过create()方法传入的接口参数,并将实际发射数据的对象回调出去，用户通过发射对象发射数据

/**
 * 被观察者
 * 提供发射数据的方法：通过create()方法传入的接口参数,将实际发射数据的对象回调出去，用户通过发射对象发射数据
 */
public class Observable<T> implements ObservableSource<T> 
    final ObservableOnSubscribe source;//类似接口监听，作用是把观察者事件发射的对象，回调出去

    private Observable(ObservableOnSubscribe source) 
        this.source = source;
    

    public static <T> Observable<T> create(ObservableOnSubscribe<T> source) 
        return new Observable(source);
    

    @Override
    public void subscribe(Observer<? super T> observer) 
        CreateEmitter<T> createEmitter = new CreateEmitter<T>(observer);
        observer.onSubscribe();
        source.subscribe(createEmitter);//把被观察者发射数据的对象回调出去
    


    //定义被观察者发射数据的对象，并把数据分发给观察者observer
    static final class CreateEmitter<T> implements ObservableEmitter<T> 
        final Observer<? super T> observer;

        public CreateEmitter(Observer<? super T> observer) 
            this.observer = observer;
        

        @Override
        public void onNext(T value) 
            observer.onNext(value);
        

        @Override
        public void onError(Throwable error) 
            onError(error);
        

        @Override
        public void onComplete() 
            observer.onComplete();
        
    

涉及到的接口ObservableOnSubscribe，这个接口的作用就是把创建的ObservableEmitter这个发射器实例对象回调出去，用户就可以使用这个发射器发射数据了。

public interface ObservableOnSubscribe<T> 
    void subscribe(ObservableEmitter<T> observableEmitter);

测试一下：

    public static void main(String[] args) 
        Observable.create(new ObservableOnSubscribe<String>() 
            @Override
            public void subscribe(ObservableEmitter<String> observableEmitter) 
                System.out.println("发射数据" + "1");
                observableEmitter.onNext("1");

            
        ).subscribe(new Observer<String>() 
            @Override
            public void onSubscribe() 

            

            @Override
            public void onNext(String value) 
                System.out.println("接收数据" + value);
            

            @Override
            public void onError(Throwable e) 

            

            @Override
            public void onComplete() 

            
        );
    

运行可以看到打印日志：-------------------------
发射数据1
接收数据1