本文共 5731 字，大约阅读时间需要 19 分钟。

一、概念

函数响应式编程(Functional Reactive Programming:FRP)：是一种通过一系列函数的组合调用来发射，转变，监听，响应数据流的编程范式（这概念我基本理解不了~）。

在RxJava中，函数响应式编程具体表现为:一个观察者（Observer） 订阅一个可观察对象（Observable）通过创建可观察对象发射数据流，经过一系列操作符（Operators）加工处理和线程调度器（Scheduler）在不同线程间的转发，最后由观察者接受并做出响应的一个过程。（来源于百度）

二、关键词

1、Observable 可观察者

2、Observer 观察者（数据接收者） 3、Subscribe 订阅 4、Emitter–>ObservableEmitter(继承了Emitter) 发射者 5、Consumer 消费者 可以看做是对观察者Observer功能单一版本

三、关联关系

1、观察者(Observer)通过订阅(Subscribe)被观察者（Observable）

2、被观察者(Observable)通过发射器（ObservableEmitter）在指定的线程中发射数据 3、观察者（Observer）在指定的线程中接收由发射器（ObservableEmitter）发射出来的数据

四、建立观察者、发送数据、接收数据的代码实现

// ObservableEmitter是事件的发送器        Observable.create(new ObservableOnSubscribe
   
    
     >() {            @Override            public void subscribe(ObservableEmitter
     
      
       > e) throws Exception {                List
       
         strList = new ArrayList<>();                for (int i = 0; i < 10; i++) {                    strList.add("test_rxjava_" + i);                }                Log.d("rxjava", "ObservableEmitter : name_id = " + Thread.currentThread().getName() + ", " + Thread.currentThread().getId());                // onError和onNext、onComplete一起发送会出错//                Throwable throwable = new Throwable("error-----------");//                e.onError(throwable);                e.onNext(strList);                e.onComplete();            }        }).subscribeOn(Schedulers.io())     // 指定observable(被观察者)运行的线程                .observeOn(Schedulers.io())  // 指定observer(观察者)运行的线程                .subscribe(new Consumer
        
         
          >() {//Observer订阅Observable 绑定observer(观察者) @Override public void accept(List
          
            strings) throws Exception { Log.d("rxjava", "accept(): name_id = " + Thread.currentThread().getName() + ", " + Thread.currentThread().getId()); if (strings != null && strings.size() > 0) { for (String str : strings) { Log.d("rxjava", "str = " + str); } } } });
          
         
        
       
      
     
    
   

注：上面的这种发送方式不支持发送错误信息，即不能执行e.onError(throwable) ， 会报错：

io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept 找不到能接收onError的消费之accept,应该用另一种三个参数的方式：

Observable.create(new ObservableOnSubscribe
   
    
     >() {            @Override            public void subscribe(ObservableEmitter
     
      
       > e) throws Exception {                List
       
         strList = getListData();                Log.d("lixm", "ObservableEmitter : name_id = " + Thread.currentThread().getName() + ", " + Thread.currentThread().getId());                // onNext可以被执行多次                e.onNext(strList);                e.onNext(strList);                // onError和onNext、onComplete一起发送会出错                // io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept                Throwable throwable = new Throwable("error-----------");                e.onError(throwable);                // 只能被执行一次                e.onComplete();            }        }).subscribeOn(Schedulers.io())     // 指定observable(被观察者)运行的线程                .observeOn(AndroidSchedulers.mainThread())  // 指定observer(观察者)运行的线程                .subscribe(new Consumer
        
         
          >() { @Override public void accept(List
          
            strings) throws Exception { Log.d("lixm", "\r\n接收到数据：accept(): name_id = " + Thread.currentThread().getName() + ", " + Thread.currentThread().getId()); if (strings != null && strings.size() > 0) { for (String str : strings) { Log.d("lixm", "strResult = " + str); } } } }, new Consumer
           
            () { @Override public void accept(Throwable throwable) throws Exception { // 出现错误信息会走到这里 Log.d("rxjava", "接收到异常信息：accept(): throwable = " + throwable.getMessage()); } }, new Action() { @Override public void run() throws Exception { // 完成以后走这里（如果走了错误的回调，就不会走这里） Log.d("rxjava", "new Action()，run(),接收到完成信息，由发射器的onComplete()方法触发."); } });
           
          
         
        
       
      
     
    
   

第三种接收方法，也是我们常用的一种方法：

public void testObservableCreate(View view) {        // 观察者        Observer
   
    
     > myObserver = new Observer
     
      
       >() {            List
       
         result = null;            @Override            public void onSubscribe(Disposable d) {                //            }            @Override            public void onNext(List
        
          strList) { Log.d("rxjava", "onNext(): name_id = " + Thread.currentThread().getName() + ", " + Thread.currentThread().getId()); if (strList != null) { result = strList; } } @Override public void onError(Throwable e) { Log.d("rxjava", "myObserver(),onError : name_id = " + Thread.currentThread().getName() + ", " + Thread.currentThread().getId()); } @Override public void onComplete() { Log.d("rxjava", "onComplete(): name_id = " + Thread.currentThread().getName() + ", " + Thread.currentThread().getId()); if (result != null && result.size() > 0) { for (String str : result) { Log.d("rxjava", "strResult = " + str); } } } }; // ObservableEmitter是事件的发送器 Observable.create(new ObservableOnSubscribe
         
          
           >() { @Override public void subscribe(ObservableEmitter
           
            
             > e) throws Exception { List
             
               strList = new ArrayList<>(); for (int i = 0; i < 10; i++) { strList.add("test_rxjava_" + i); } Log.d("rxjava", "ObservableEmitter : name_id = " + Thread.currentThread().getName() + ", " + Thread.currentThread().getId());// Throwable throwable = new Throwable("error-----------");// e.onError(throwable); e.onNext(strList); e.onComplete(); } }).subscribeOn(Schedulers.io()) // 指定observable(被观察者)运行的线程 .observeOn(Schedulers.io()) // 指定observer(观察者)运行的线程 .subscribe(myObserver); //Observer订阅Observable 绑定observer(观察者) }
             
            
           
          
         
        
       
      
     
    
   

五、分析一下发射器和接受者的关系

1、ObservableEmitter 继承自 Emitter

2、发射器中的三个方法对应接收器中的三个方法 即 ：Emitter–> void onNext(@NonNull T value); –> void onError(@NonNull Throwable error); –> void onComplete(); Observer --> void onNext(@NonNull T t); –> void onError(@NonNull Throwable e); –>void onComplete();

转载地址：https://blog.csdn.net/dami_lixm/article/details/96480642 如侵犯您的版权，请留言回复原文章的地址，我们会给您删除此文章，给您带来不便请您谅解！