devbox
我家自动化
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

RxJava详解(基于2.X版本的背压详解)(下)_io.reactivex.android.schedulers.androidschedulers

作者：我家自动化 | 2024-03-24 07:40:18

赞

踩

io.reactivex.android.schedulers.androidschedulers

RxJava背压基础讲解

https://blog.csdn.net/weixin_37730482/article/details/78085190

 

 

本章节继续讲述RxJava的背压问题

 

 

 

 

 

一.控制 观察者 接收事件的速度

 

<1> 前言

 

当被观察者发送消息的速度和观察者接收消息的速度不匹配时，可以控制观察者接收消息的速度。


原理是 不管被观察者发送了多少条数据，观察者仅根据自身情况选择接收几条消息。哪怕被观察者发送量100条数据，若观察者仅需要两条消息，则只接收两条消息。忽略另外98条消息。

 

 

 

<2> 代码

  1. package com.example.mydemo.rxjava;
  2.  
  3. import android.os.Bundle;
  4. import android.util.Log;
  5.  
  6. import androidx.appcompat.app.AppCompatActivity;
  7.  
  8. import com.example.mydemo.R;
  9.  
  10. import org.reactivestreams.Subscriber;
  11. import org.reactivestreams.Subscription;
  12.  
  13. import io.reactivex.BackpressureStrategy;
  14. import io.reactivex.Flowable;
  15. import io.reactivex.FlowableEmitter;
  16. import io.reactivex.FlowableOnSubscribe;
  17. import io.reactivex.android.schedulers.AndroidSchedulers;
  18. import io.reactivex.schedulers.Schedulers;
  19.  
  20. public class RxJavaActivity extends AppCompatActivity {
  21.  
  22.     private Subscription mSubscription;
  23.  
  24.     @Override
  25.     protected void onCreate(Bundle savedInstanceState) {
  26.         super.onCreate(savedInstanceState);
  27.         setContentView(R.layout.activity_rxjava);
  28.  
  29.         method();
  30.     }
  31.  
  32.     /**
  33.      * RxJava Flowable创建被观察者 背压
  34.      */
  35.  
  36.     public void method() {
  37.  
  38.         Flowable.create(new FlowableOnSubscribe<Object>() {
  39.             @Override
  40.             public void subscribe(FlowableEmitter<Object> e) throws Exception {
  41.                 if (null == e) {
  42.                     return;
  43.                 }
  44.  
  45.                 /**
  46.                  * 测试 被观察者发送五条消息
  47.                  * */
  48.  
  49.                 if (!e.isCancelled()) {
  50.                     for (int i = 0; i < 5; i++) {
  51.                         e.onNext("Flowable创建被观察者 被观察者发送第   " + i + "   波数据");
  52.                     }
  53.                     e.onComplete();
  54.                 }
  55.             }
  56.         }, BackpressureStrategy.LATEST)
  57.                 .subscribeOn(Schedulers.io())//被观察者在io线程中进行
  58.                 .observeOn(AndroidSchedulers.mainThread())//观察者在主线程中进行
  59.                 .subscribe(new Subscriber<Object>() {
  60.                     @Override
  61.                     public void onSubscribe(Subscription s) {
  62.                         if (null == s) {
  63.                             return;
  64.                         }
  65.                         mSubscription = s;
  66.                         mSubscription.request(4);//测试 被观察者发送五条消息 观察者只接收四条
  67.                     }
  68.  
  69.                     @Override
  70.                     public void onNext(Object o) {
  71.                         if (null == o) {
  72.                             return;
  73.                         }
  74.                         Log.d("TAG", "Flowable创建被观察者 观察者 onNext方法 o.toString()----:" + o.toString());
  75.                     }
  76.  
  77.                     @Override
  78.                     public void onError(Throwable t) {
  79.                         if (null == t) {
  80.                             return;
  81.                         }
  82.                         Log.d("TAG", "Flowable创建被观察者 观察者 onError方法 t.toString()----:" + t.toString());
  83.                     }
  84.  
  85.                     @Override
  86.                     public void onComplete() {
  87.                         Log.d("TAG", "Flowable创建被观察者 观察者 onComplete方法 ");
  88.                     }
  89.                 });
  90.     }
  91.  
  92.     /**
  93.      * onDestroy方法 取消订阅
  94.      */
  95.  
  96.     @Override
  97.     protected void onDestroy() {
  98.         super.onDestroy();
  99.         if (null != mSubscription) {
  100.             mSubscription.cancel();
  101.             Log.d("TAG", "退出页面 取消订阅");
  102.         }
  103.     }
  104.  
  105. }

 

 

 

<3> 结果

  1. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   0   波数据
  2.  
  3.  
  4. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   1   波数据
  5.  
  6. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   2   波数据
  7.  
  8. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   3   波数据

 

 

<4> 说明

被观察者发送五条消息。观察者通过onSubscribe方法中参数Subscription的request(number)方法设置只接收number即四条消息。

 

 

 

<5> 重要说明&异步订阅

Flowable抽象类创建被观察者方式  如果被观察者和观察者是异步订阅的 如果不设置

mSubscription.request(N);//设置 观察者只接收N条数据

则说明 观察者不接收消息。此时 被观察者仍然可以发送消息。并将消息存在缓存区。等待观察者需要时再取出。即调用

mSubscription.request(N);//设置 观察者只接收N条数据

方法。

 

(1) 代码

  1. package com.example.mydemo.rxjava;
  2.  
  3. import android.os.Bundle;
  4. import android.util.Log;
  5. import android.view.View;
  6. import android.widget.TextView;
  7.  
  8. import androidx.appcompat.app.AppCompatActivity;
  9.  
  10. import com.example.mydemo.R;
  11.  
  12. import org.reactivestreams.Subscriber;
  13. import org.reactivestreams.Subscription;
  14.  
  15. import io.reactivex.BackpressureStrategy;
  16. import io.reactivex.Flowable;
  17. import io.reactivex.FlowableEmitter;
  18. import io.reactivex.FlowableOnSubscribe;
  19. import io.reactivex.android.schedulers.AndroidSchedulers;
  20. import io.reactivex.schedulers.Schedulers;
  21.  
  22. public class RxJavaActivity extends AppCompatActivity {
  23.  
  24.     private Subscription mSubscription;
  25.  
  26.     @Override
  27.     protected void onCreate(Bundle savedInstanceState) {
  28.         super.onCreate(savedInstanceState);
  29.         setContentView(R.layout.activity_rxjava);
  30.         findView();
  31.         method();
  32.     }
  33.  
  34.     /**
  35.      * 初始化各种View
  36.      */
  37.  
  38.     private void findView() {
  39.         TextView textView = findViewById(R.id.activity_rxjava_textview);
  40.         textView.setOnClickListener(new View.OnClickListener() {
  41.             @Override
  42.             public void onClick(View v) {
  43.                 if (null != mSubscription) {
  44.                     mSubscription.request(4);//观察者只接收四条
  45.                 }
  46.             }
  47.         });
  48.     }
  49.  
  50.     /**
  51.      * RxJava Flowable创建被观察者 背压
  52.      */
  53.  
  54.     public void method() {
  55.  
  56.         Flowable.create(new FlowableOnSubscribe<Object>() {
  57.             @Override
  58.             public void subscribe(FlowableEmitter<Object> e) throws Exception {
  59.                 if (null == e) {
  60.                     return;
  61.                 }
  62.  
  63.                 /**
  64.                  * 测试 被观察者发送五条消息
  65.                  * */
  66.  
  67.                 if (!e.isCancelled()) {
  68.                     for (int i = 0; i < 5; i++) {
  69.                         e.onNext("Flowable创建被观察者 被观察者发送第   " + i + "   波数据");
  70.                     }
  71.                     e.onComplete();
  72.                 }
  73.             }
  74.         }, BackpressureStrategy.LATEST)
  75.                 .subscribeOn(Schedulers.io())//被观察者在io线程中进行
  76.                 .observeOn(AndroidSchedulers.mainThread())//观察者在主线程中进行
  77.                 .subscribe(new Subscriber<Object>() {
  78.                     @Override
  79.                     public void onSubscribe(Subscription s) {
  80.                         if (null == s) {
  81.                             return;
  82.                         }
  83.                         mSubscription = s;
  84.                     }
  85.  
  86.                     @Override
  87.                     public void onNext(Object o) {
  88.                         if (null == o) {
  89.                             return;
  90.                         }
  91.                         Log.d("TAG", "Flowable创建被观察者 观察者 onNext方法 o.toString()----:" + o.toString());
  92.                     }
  93.  
  94.                     @Override
  95.                     public void onError(Throwable t) {
  96.                         if (null == t) {
  97.                             return;
  98.                         }
  99.                         Log.d("TAG", "Flowable创建被观察者 观察者 onError方法 t.toString()----:" + t.toString());
  100.                     }
  101.  
  102.                     @Override
  103.                     public void onComplete() {
  104.                         Log.d("TAG", "Flowable创建被观察者 观察者 onComplete方法 ");
  105.                     }
  106.                 });
  107.     }
  108.  
  109.     /**
  110.      * onDestroy方法 取消订阅
  111.      */
  112.  
  113.     @Override
  114.     protected void onDestroy() {
  115.         super.onDestroy();
  116.         if (null != mSubscription) {
  117.             mSubscription.cancel();
  118.             Log.d("TAG", "退出页面 取消订阅");
  119.         }
  120.     }
  121.  
  122. }

 

(2) 结果

默认

无打印 即无消息接收

 

点击按钮 点击第一次 取出四条消息。

  1. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   0   波数据
  2.  
  3. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   1   波数据
  4.  
  5. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   2   波数据
  6.  
  7. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   3   波数据

 

点击按钮 点击第二次 取出剩余消息并执行onComplete方法。

  1. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   4   波数据
  2.  
  3. D/TAG: Flowable创建被观察者 观察者 onComplete方法

 

点击按钮 点击第三四五...次 

无打印 即无消息接收。

 

 

 

 

<6> 重要说明&同步订阅

 

同步订阅，即被观察者和观察者处在同一线程中。同步订阅关系中没有缓存区。

 

同步订阅&异步订阅区别

订阅关系说明
同步订阅观察者和被观察者处在同一线程。被观察者每发送一条消息。必须等到观察者接收后才会发送下一条消息。
异步订阅观察者和被观察者处在不同的线程。被观察者不需要等待观察者处理消息后再发送下一条消息，而是不断发送直到发送消息完毕。

 

 

 

 

 

 

 

 代码

 

[1] 被观察者发送的消息数和观察者设置接收的消息数一致时。

 

(1) 代码

  1. package com.example.mydemo.rxjava;
  2.  
  3. import android.os.Bundle;
  4. import android.util.Log;
  5.  
  6. import androidx.appcompat.app.AppCompatActivity;
  7.  
  8. import com.example.mydemo.R;
  9.  
  10. import org.reactivestreams.Subscriber;
  11. import org.reactivestreams.Subscription;
  12.  
  13. import io.reactivex.BackpressureStrategy;
  14. import io.reactivex.Flowable;
  15. import io.reactivex.FlowableEmitter;
  16. import io.reactivex.FlowableOnSubscribe;
  17.  
  18. public class RxJavaActivity extends AppCompatActivity {
  19.  
  20.     private Subscription mSubscription;
  21.  
  22.     @Override
  23.     protected void onCreate(Bundle savedInstanceState) {
  24.         super.onCreate(savedInstanceState);
  25.         setContentView(R.layout.activity_rxjava);
  26.         method();
  27.     }
  28.  
  29.     /**
  30.      * RxJava Flowable创建被观察者 背压
  31.      */
  32.  
  33.     public void method() {
  34.  
  35.         Flowable.create(new FlowableOnSubscribe<Object>() {
  36.             @Override
  37.             public void subscribe(FlowableEmitter<Object> e) throws Exception {
  38.                 if (null == e) {
  39.                     return;
  40.                 }
  41.  
  42.                 /**
  43.                  * 测试 被观察者发送五条消息
  44.                  * */
  45.  
  46.                 if (!e.isCancelled()) {
  47.                     for (int i = 0; i < 5; i++) {
  48.                         e.onNext("Flowable创建被观察者 被观察者发送第   " + i + "   波数据");
  49.                         Log.d("TAG", "Flowable创建被观察者 被观察者发送数据 i----：" + i);
  50.                     }
  51.                     e.onComplete();
  52.                 }
  53.             }
  54.         }, BackpressureStrategy.LATEST)
  55.                 .subscribe(new Subscriber<Object>() {
  56.                     @Override
  57.                     public void onSubscribe(Subscription s) {
  58.                         if (null == s) {
  59.                             return;
  60.                         }
  61.                         mSubscription = s;
  62.                         mSubscription.request(5);//观察者接收五条
  63.                     }
  64.  
  65.                     @Override
  66.                     public void onNext(Object o) {
  67.                         if (null == o) {
  68.                             return;
  69.                         }
  70.                         Log.d("TAG", "Flowable创建被观察者 观察者 onNext方法 o.toString()----:" + o.toString());
  71.                     }
  72.  
  73.                     @Override
  74.                     public void onError(Throwable t) {
  75.                         if (null == t) {
  76.                             return;
  77.                         }
  78.                         Log.d("TAG", "Flowable创建被观察者 观察者 onError方法 t.toString()----:" + t.toString());
  79.                     }
  80.  
  81.                     @Override
  82.                     public void onComplete() {
  83.                         Log.d("TAG", "Flowable创建被观察者 观察者 onComplete方法 ");
  84.                     }
  85.                 });
  86.     }
  87.  
  88.     /**
  89.      * onDestroy方法 取消订阅
  90.      */
  91.  
  92.     @Override
  93.     protected void onDestroy() {
  94.         super.onDestroy();
  95.         if (null != mSubscription) {
  96.             mSubscription.cancel();
  97.             Log.d("TAG", "退出页面 取消订阅");
  98.         }
  99.     }
  100.  
  101. }

 

 

(2) 结果

  1. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   0   波数据
  2.  
  3. D/TAG: Flowable创建被观察者 被观察者发送数据 i----：0
  4.  
  5.  
  6.  
  7.  
  8. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   1   波数据
  9.  
  10. D/TAG: Flowable创建被观察者 被观察者发送数据 i----：1
  11.  
  12.  
  13.  
  14.  
  15. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   2   波数据
  16.  
  17. D/TAG: Flowable创建被观察者 被观察者发送数据 i----：2
  18.  
  19.  
  20.  
  21. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   3   波数据
  22.  
  23. D/TAG: Flowable创建被观察者 被观察者发送数据 i----：3
  24.  
  25.  
  26.  
  27. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   4   波数据
  28.  
  29. D/TAG: Flowable创建被观察者 被观察者发送数据 i----：4
  30.  
  31.  
  32.  
  33. D/TAG: Flowable创建被观察者 观察者 onComplete方法

 

(3) 综上所述，同步订阅情况下不会出现被观察者发送事件速度 > 观察者接收事件速度的情况。可是，却会出现被观察者发送事件数量 > 观察者接收事件数量的问题。

 

 

[2] 被观察者发送的消息数和观察者设置接收的消息数不一致时。

 

(1) 代码

  1. package com.example.mydemo.rxjava;
  2.  
  3. import android.os.Bundle;
  4. import android.util.Log;
  5.  
  6. import androidx.appcompat.app.AppCompatActivity;
  7.  
  8. import com.example.mydemo.R;
  9.  
  10. import org.reactivestreams.Subscriber;
  11. import org.reactivestreams.Subscription;
  12.  
  13. import io.reactivex.BackpressureStrategy;
  14. import io.reactivex.Flowable;
  15. import io.reactivex.FlowableEmitter;
  16. import io.reactivex.FlowableOnSubscribe;
  17.  
  18. public class RxJavaActivity extends AppCompatActivity {
  19.  
  20.     private Subscription mSubscription;
  21.  
  22.     @Override
  23.     protected void onCreate(Bundle savedInstanceState) {
  24.         super.onCreate(savedInstanceState);
  25.         setContentView(R.layout.activity_rxjava);
  26.         method();
  27.     }
  28.  
  29.     /**
  30.      * RxJava Flowable创建被观察者 背压
  31.      */
  32.  
  33.     public void method() {
  34.  
  35.         Flowable.create(new FlowableOnSubscribe<Object>() {
  36.             @Override
  37.             public void subscribe(FlowableEmitter<Object> e) throws Exception {
  38.                 if (null == e) {
  39.                     return;
  40.                 }
  41.  
  42.                 /**
  43.                  * 测试 被观察者发送五条消息
  44.                  * */
  45.  
  46.                 if (!e.isCancelled()) {
  47.                     for (int i = 0; i < 5; i++) {
  48.                         e.onNext("Flowable创建被观察者 被观察者发送第   " + i + "   波数据");
  49.                         Log.d("TAG", "Flowable创建被观察者 被观察者发送数据 i----：" + i);
  50.                     }
  51.                     e.onComplete();
  52.                 }
  53.             }
  54.         }, BackpressureStrategy.LATEST)
  55.                 .subscribe(new Subscriber<Object>() {
  56.                     @Override
  57.                     public void onSubscribe(Subscription s) {
  58.                         if (null == s) {
  59.                             return;
  60.                         }
  61.                         mSubscription = s;
  62.                         mSubscription.request(3);//观察者接收三条
  63.                     }
  64.  
  65.                     @Override
  66.                     public void onNext(Object o) {
  67.                         if (null == o) {
  68.                             return;
  69.                         }
  70.                         Log.d("TAG", "Flowable创建被观察者 观察者 onNext方法 o.toString()----:" + o.toString());
  71.                     }
  72.  
  73.                     @Override
  74.                     public void onError(Throwable t) {
  75.                         if (null == t) {
  76.                             return;
  77.                         }
  78.                         Log.d("TAG", "Flowable创建被观察者 观察者 onError方法 t.toString()----:" + t.toString());
  79.                     }
  80.  
  81.                     @Override
  82.                     public void onComplete() {
  83.                         Log.d("TAG", "Flowable创建被观察者 观察者 onComplete方法 ");
  84.                     }
  85.                 });
  86.     }
  87.  
  88.     /**
  89.      * onDestroy方法 取消订阅
  90.      */
  91.  
  92.     @Override
  93.     protected void onDestroy() {
  94.         super.onDestroy();
  95.         if (null != mSubscription) {
  96.             mSubscription.cancel();
  97.             Log.d("TAG", "退出页面 取消订阅");
  98.         }
  99.     }
  100.  
  101. }

即 被观察者发送五条数据 观察者只接收三条消息。

 

 

(2) 结果

  1. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   0   波数据
  2.  
  3. D/TAG: Flowable创建被观察者 被观察者发送数据 i----：0
  4.  
  5.  
  6.  
  7. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   1   波数据
  8.  
  9. D/TAG: Flowable创建被观察者 被观察者发送数据 i----：1
  10.  
  11.  
  12.  
  13. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   2   波数据
  14.  
  15. D/TAG: Flowable创建被观察者 被观察者发送数据 i----：2
  16.  
  17.  
  18.  
  19. D/TAG: Flowable创建被观察者 被观察者发送数据 i----：3
  20.  
  21.  
  22.  
  23. D/TAG: Flowable创建被观察者 被观察者发送数据 i----：4

 

(3) 综上所述 对于没有缓存区概念的同步订阅关系来说，单纯采用控制观察者的接收事件数量（响应式拉取）还是会出现问题。

 

 

 

[3] 点击时再执行观察者的取方法

 

(1) 代码

  1. package com.example.mydemo.rxjava;
  2.  
  3. import android.os.Bundle;
  4. import android.util.Log;
  5. import android.view.View;
  6. import android.widget.TextView;
  7.  
  8. import androidx.appcompat.app.AppCompatActivity;
  9.  
  10. import com.example.mydemo.R;
  11.  
  12. import org.reactivestreams.Subscriber;
  13. import org.reactivestreams.Subscription;
  14.  
  15. import io.reactivex.BackpressureStrategy;
  16. import io.reactivex.Flowable;
  17. import io.reactivex.FlowableEmitter;
  18. import io.reactivex.FlowableOnSubscribe;
  19.  
  20. public class RxJavaActivity extends AppCompatActivity {
  21.  
  22.     private Subscription mSubscription;
  23.  
  24.     @Override
  25.     protected void onCreate(Bundle savedInstanceState) {
  26.         super.onCreate(savedInstanceState);
  27.         setContentView(R.layout.activity_rxjava);
  28.         findView();
  29.         method();
  30.     }
  31.  
  32.     /**
  33.      * 初始化各种View
  34.      */
  35.  
  36.     private void findView() {
  37.         TextView textView = findViewById(R.id.activity_rxjava_textview);
  38.         textView.setOnClickListener(new View.OnClickListener() {
  39.             @Override
  40.             public void onClick(View v) {
  41.                 if (null != mSubscription) {
  42.                     mSubscription.request(4);//观察者只接收四条
  43.                 }
  44.             }
  45.         });
  46.     }
  47.  
  48.     /**
  49.      * RxJava Flowable创建被观察者 背压
  50.      */
  51.  
  52.     public void method() {
  53.  
  54.         Flowable.create(new FlowableOnSubscribe<Object>() {
  55.             @Override
  56.             public void subscribe(FlowableEmitter<Object> e) throws Exception {
  57.                 if (null == e) {
  58.                     return;
  59.                 }
  60.  
  61.                 /**
  62.                  * 测试 被观察者发送五条消息
  63.                  * */
  64.  
  65.                 if (!e.isCancelled()) {
  66.                     for (int i = 0; i < 5; i++) {
  67.                         e.onNext("Flowable创建被观察者 被观察者发送第   " + i + "   波数据");
  68.                     }
  69.                     e.onComplete();
  70.                 }
  71.             }
  72.         }, BackpressureStrategy.LATEST)
  73.                 .subscribe(new Subscriber<Object>() {
  74.                     @Override
  75.                     public void onSubscribe(Subscription s) {
  76.                         if (null == s) {
  77.                             return;
  78.                         }
  79.                         mSubscription = s;
  80.                     }
  81.  
  82.                     @Override
  83.                     public void onNext(Object o) {
  84.                         if (null == o) {
  85.                             return;
  86.                         }
  87.                         Log.d("TAG", "Flowable创建被观察者 观察者 onNext方法 o.toString()----:" + o.toString());
  88.                     }
  89.  
  90.                     @Override
  91.                     public void onError(Throwable t) {
  92.                         if (null == t) {
  93.                             return;
  94.                         }
  95.                         Log.d("TAG", "Flowable创建被观察者 观察者 onError方法 t.toString()----:" + t.toString());
  96.                     }
  97.  
  98.                     @Override
  99.                     public void onComplete() {
  100.                         Log.d("TAG", "Flowable创建被观察者 观察者 onComplete方法 ");
  101.                     }
  102.                 });
  103.     }
  104.  
  105.     /**
  106.      * onDestroy方法 取消订阅
  107.      */
  108.  
  109.     @Override
  110.     protected void onDestroy() {
  111.         super.onDestroy();
  112.         if (null != mSubscription) {
  113.             mSubscription.cancel();
  114.             Log.d("TAG", "退出页面 取消订阅");
  115.         }
  116.     }
  117.  
  118. }

 

 

(2) 结果

 

默认

无打印 即无消息接收

 

点击按钮 第一次点击

  1. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   4   波数据
  2.  
  3. D/TAG: Flowable创建被观察者 观察者 onComplete方法

 

点击按钮第二三四五...次点击

无打印 即无消息接收

 

 

 

 

 

 

 

二.控制 被观察者 发送事件的速度

 

<1> 前言

上述可知可以通过观察者的onSubscribe方法中参数Subscription的request(N)方法控制观察者可接收的消息数，继而控制观察者接收事件的速度。那么被观察者的发送事件的速度怎么控制呢？

 

源码分析

 

create操作符创建被观察者 重写 subscribe方法 该方法参数FlowableEmitter<Object> e 点击

  1. public interface FlowableEmitter<T> extends Emitter<T> {
  2.  
  3.     /**
  4.      * Sets a Disposable on this emitter; any previous Disposable
  5.      * or Cancellation will be unsubscribed/cancelled.
  6.      * @param s the disposable, null is allowed
  7.      */
  8.     void setDisposable(Disposable s);
  9.  
  10.     /**
  11.      * Sets a Cancellable on this emitter; any previous Disposable
  12.      * or Cancellation will be unsubscribed/cancelled.
  13.      * @param c the cancellable resource, null is allowed
  14.      */
  15.     void setCancellable(Cancellable c);
  16.  
  17.     /**
  18.      * The current outstanding request amount.
  19.      * <p>This method is thread-safe.
  20.      * @return the current outstanding request amount
  21.      */
  22.     long requested();
  23.  
  24.     /**
  25.      * Returns true if the downstream cancelled the sequence.
  26.      * <p>This method is thread-safe.
  27.      * @return true if the downstream cancelled the sequence
  28.      */
  29.     boolean isCancelled();
  30.  
  31.     /**
  32.      * Ensures that calls to onNext, onError and onComplete are properly serialized.
  33.      * @return the serialized FlowableEmitter
  34.      */
  35.     FlowableEmitter<T> serialize();
  36. }

 

可以看到 此参数可以判断是否取消订阅  isCancelled()方法。而request()方法可以获取被观察者中的消息数量。

 

 

<2> 同步订阅 控制被观察者发送消息速度

 

(1) 简介

同步订阅中，被观察者和观察者处在同一线程。被观察者中request方法得到的数等于观察者中Subscription.request()方法中设置的数。

 

 

(2) 代码

  1. package com.example.mydemo.rxjava;
  2.  
  3. import android.os.Bundle;
  4. import android.util.Log;
  5.  
  6. import androidx.appcompat.app.AppCompatActivity;
  7.  
  8. import com.example.mydemo.R;
  9.  
  10. import org.reactivestreams.Subscriber;
  11. import org.reactivestreams.Subscription;
  12.  
  13. import io.reactivex.BackpressureStrategy;
  14. import io.reactivex.Flowable;
  15. import io.reactivex.FlowableEmitter;
  16. import io.reactivex.FlowableOnSubscribe;
  17.  
  18. public class RxJavaActivity extends AppCompatActivity {
  19.  
  20.     private Subscription mSubscription;
  21.  
  22.     @Override
  23.     protected void onCreate(Bundle savedInstanceState) {
  24.         super.onCreate(savedInstanceState);
  25.         setContentView(R.layout.activity_rxjava);
  26.         method();
  27.     }
  28.  
  29.     /**
  30.      * RxJava Flowable创建被观察者 背压
  31.      */
  32.  
  33.     public void method() {
  34.  
  35.         Flowable.create(new FlowableOnSubscribe<Object>() {
  36.             @Override
  37.             public void subscribe(FlowableEmitter<Object> e) throws Exception {
  38.                 if (null == e) {
  39.                     return;
  40.                 }
  41.  
  42.                 long msgNumber = e.requested();
  43.                 Log.d("TAG", "Flowable创建被观察者 被观察者 msgNumber----:" + msgNumber);
  44.  
  45.                 if (!e.isCancelled()) {
  46.                     for (int i = 0; i < msgNumber; i++) {
  47.                         e.onNext("Flowable创建被观察者 被观察者发送第   " + i + "   波数据");
  48.                     }
  49.                     e.onComplete();
  50.                 }
  51.             }
  52.         }, BackpressureStrategy.LATEST)
  53.                 .subscribe(new Subscriber<Object>() {
  54.                     @Override
  55.                     public void onSubscribe(Subscription s) {
  56.                         if (null == s) {
  57.                             return;
  58.                         }
  59.                         mSubscription = s;
  60.                         mSubscription.request(5);//观察者设置可接收五条消息
  61.                     }
  62.  
  63.                     @Override
  64.                     public void onNext(Object o) {
  65.                         if (null == o) {
  66.                             return;
  67.                         }
  68.                         Log.d("TAG", "Flowable创建被观察者 观察者 onNext方法 o.toString()----:" + o.toString());
  69.                     }
  70.  
  71.                     @Override
  72.                     public void onError(Throwable t) {
  73.                         if (null == t) {
  74.                             return;
  75.                         }
  76.                         Log.d("TAG", "Flowable创建被观察者 观察者 onError方法 t.toString()----:" + t.toString());
  77.                     }
  78.  
  79.                     @Override
  80.                     public void onComplete() {
  81.                         Log.d("TAG", "Flowable创建被观察者 观察者 onComplete方法 ");
  82.                     }
  83.                 });
  84.     }
  85.  
  86.     /**
  87.      * onDestroy方法 取消订阅
  88.      */
  89.  
  90.     @Override
  91.     protected void onDestroy() {
  92.         super.onDestroy();
  93.         if (null != mSubscription) {
  94.             mSubscription.cancel();
  95.             Log.d("TAG", "退出页面 取消订阅");
  96.         }
  97.     }
  98.  
  99. }

 

 

(3) 结果

  1. D/TAG: Flowable创建被观察者 被观察者 msgNumber----:5
  2.  
  3.  
  4. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   0   波数据
  5.  
  6. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   1   波数据
  7.  
  8. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   2   波数据
  9.  
  10. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   3   波数据
  11.  
  12. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   4   波数据
  13.  
  14. D/TAG: Flowable创建被观察者 观察者 onComplete方法

 

 

(4) 说明

{1} 被观察者的 e.requested();方法具有实时性。即被观察者发送了一条数据 e.requested();方法就会减一。

 

代码 在上基础上改动

  1. long msgNumber = e.requested();
  2. Log.d("TAG", "Flowable创建被观察者 被观察者 msgNumber----:" + msgNumber);
  3.  
  4. if (!e.isCancelled()) {
  5.     for (int i = 0; i < msgNumber; i++) {
  6.         e.onNext("Flowable创建被观察者 被观察者发送第   " + i + "   波数据");
  7.         Log.d("TAG", "Flowable创建被观察者 被观察者 发送了i后e.requested()----:" + e.requested());
  8.     }
  9.     e.onComplete();
  10. }

 

结果

  1. D/TAG: Flowable创建被观察者 被观察者 msgNumber----:5
  2.  
  3. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   0   波数据
  4.  
  5. D/TAG: Flowable创建被观察者 被观察者 发送了i后e.requested()----:4
  6.  
  7. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   1   波数据
  8.  
  9. D/TAG: Flowable创建被观察者 被观察者 发送了i后e.requested()----:3
  10.  
  11. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   2   波数据
  12.  
  13. D/TAG: Flowable创建被观察者 被观察者 发送了i后e.requested()----:2
  14.  
  15. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   3   波数据
  16.  
  17. D/TAG: Flowable创建被观察者 被观察者 发送了i后e.requested()----:1
  18.  
  19. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   4   波数据
  20.  
  21. D/TAG: Flowable创建被观察者 被观察者 发送了i后e.requested()----:0
  22.  
  23. D/TAG: Flowable创建被观察者 观察者 onComplete方法

 

{2} 观察者没有设置可接收事件数量，即无调用Subscription.request()。那么被观察者默认观察者可接收事件数量 = 0，即FlowableEmitter.requested()的返回值 = 0。

 

{3} 可叠加性

 

代码

  1. mSubscription.request(5);//第一次设置 观察者设置可接收五条消息 
  2.  
  3. mSubscription.request(10);//第二条设置 观察者设置可接收十条消息

 

结果 5+10=15

D/TAG: Flowable创建被观察者 被观察者 msgNumber----:15

 

 

 

 

<3> 异步订阅 控制被观察者发送消息速度

 

(1) 说明

异步订阅中，被观察者和观察者处在不同的线程。所以被观察者 无法通过 FlowableEmitter.requested()知道观察者自身接收事件能力。即 被观察者不能根据 观察者自身接收事件的能力 控制发送事件的速度。

被观察者中request方法得到的数==128，96，0。 和观察者中Subscription.request()方法中设置的数不相等。

 

 

(2) 代码

  1. package com.example.mydemo.rxjava;
  2.  
  3. import android.os.Bundle;
  4. import android.util.Log;
  5.  
  6. import androidx.appcompat.app.AppCompatActivity;
  7.  
  8. import com.example.mydemo.R;
  9.  
  10. import org.reactivestreams.Subscriber;
  11. import org.reactivestreams.Subscription;
  12.  
  13. import io.reactivex.BackpressureStrategy;
  14. import io.reactivex.Flowable;
  15. import io.reactivex.FlowableEmitter;
  16. import io.reactivex.FlowableOnSubscribe;
  17. import io.reactivex.android.schedulers.AndroidSchedulers;
  18. import io.reactivex.schedulers.Schedulers;
  19.  
  20. public class RxJavaActivity extends AppCompatActivity {
  21.  
  22.     private Subscription mSubscription;
  23.  
  24.     @Override
  25.     protected void onCreate(Bundle savedInstanceState) {
  26.         super.onCreate(savedInstanceState);
  27.         setContentView(R.layout.activity_rxjava);
  28.         method();
  29.     }
  30.  
  31.     /**
  32.      * RxJava Flowable创建被观察者 背压
  33.      */
  34.  
  35.     public void method() {
  36.  
  37.         Flowable.create(new FlowableOnSubscribe<Object>() {
  38.             @Override
  39.             public void subscribe(FlowableEmitter<Object> e) throws Exception {
  40.                 if (null == e) {
  41.                     return;
  42.                 }
  43.  
  44.                 long msgNumber = e.requested();
  45.                 Log.d("TAG", "Flowable创建被观察者 被观察者 msgNumber----:" + msgNumber);
  46.  
  47.                 if (!e.isCancelled()) {
  48.                     for (int i = 0; i < msgNumber; i++) {
  49.                         e.onNext("Flowable创建被观察者 被观察者发送第   " + i + "   波数据");
  50.                     }
  51.                     e.onComplete();
  52.                 }
  53.             }
  54.         }, BackpressureStrategy.LATEST)
  55.                 .subscribeOn(Schedulers.io())//被观察者 IO线程
  56.                 .observeOn(AndroidSchedulers.mainThread())//观察者 UI线程
  57.                 .subscribe(new Subscriber<Object>() {
  58.                     @Override
  59.                     public void onSubscribe(Subscription s) {
  60.                         if (null == s) {
  61.                             return;
  62.                         }
  63.                         mSubscription = s;
  64.                         mSubscription.request(5);//观察者设置可接收五条消息
  65.                     }
  66.  
  67.                     @Override
  68.                     public void onNext(Object o) {
  69.                         if (null == o) {
  70.                             return;
  71.                         }
  72.                         Log.d("TAG", "Flowable创建被观察者 观察者 onNext方法 o.toString()----:" + o.toString());
  73.                     }
  74.  
  75.                     @Override
  76.                     public void onError(Throwable t) {
  77.                         if (null == t) {
  78.                             return;
  79.                         }
  80.                         Log.d("TAG", "Flowable创建被观察者 观察者 onError方法 t.toString()----:" + t.toString());
  81.                     }
  82.  
  83.                     @Override
  84.                     public void onComplete() {
  85.                         Log.d("TAG", "Flowable创建被观察者 观察者 onComplete方法 ");
  86.                     }
  87.                 });
  88.     }
  89.  
  90.     /**
  91.      * onDestroy方法 取消订阅
  92.      */
  93.  
  94.     @Override
  95.     protected void onDestroy() {
  96.         super.onDestroy();
  97.         if (null != mSubscription) {
  98.             mSubscription.cancel();
  99.             Log.d("TAG", "退出页面 取消订阅");
  100.         }
  101.     }
  102.  
  103. }

 

 

(3) 结果

  1. D/TAG: Flowable创建被观察者 被观察者 msgNumber----:128
  2.  
  3.  
  4. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   0   波数据
  5.  
  6. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   1   波数据
  7.  
  8. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   2   波数据
  9.  
  10. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   3   波数据
  11.  
  12. D/TAG: Flowable创建被观察者 观察者 onNext方法 o.toString()----:Flowable创建被观察者 被观察者发送第   4   波数据

即 观察者设置可以接收的消息数是五条，但是被观察者中请求的消息数是128。

 

 

 

 

 

 

 

三.Flowable其他操作符使用背压策略

 

<1> 前言

上述的讲解都是使用的Flowable的create操作符创建的被观察者，那么背压策略可以在第二个参数传入。如下。

  1. Flowable.create(new FlowableOnSubscribe<Object>() {
  2.     @Override
  3.     public void subscribe(FlowableEmitter<Object> e) throws Exception {
  4.         if (null == e) {
  5.             return;
  6.         }
  7.  
  8.         if (!e.isCancelled()) {
  9.             e.onNext("Flowable创建被观察者 被观察者发送第1波数据");
  10.             e.onNext("Flowable创建被观察者 被观察者发送第2波数据");
  11.             e.onNext("Flowable创建被观察者 被观察者发送第3波数据");
  12.             e.onComplete();
  13.         }
  14.  
  15.         Log.d("TAG", "Flowable创建被观察者 被观察者 subscribe方法 线程----：" + Thread.currentThread().getName());
  16.     }
  17. }, BackpressureStrategy.LATEST);

 

那么使用其他的操作符呢，比如前节讲的interval操作符。下面讲解。

 

 

 

<2> 代码

  1. package com.wjn.rxdemo.rxjava;
  2.  
  3. import android.os.Bundle;
  4. import android.util.Log;
  5.  
  6. import androidx.appcompat.app.AppCompatActivity;
  7.  
  8. import com.wjn.rxdemo.R;
  9.  
  10. import org.reactivestreams.Subscriber;
  11. import org.reactivestreams.Subscription;
  12.  
  13. import java.util.concurrent.TimeUnit;
  14.  
  15. import io.reactivex.Flowable;
  16. import io.reactivex.schedulers.Schedulers;
  17.  
  18. public class RxJavaActivity extends AppCompatActivity {
  19.  
  20.     private Subscription mSubscription;
  21.  
  22.     @Override
  23.     protected void onCreate(Bundle savedInstanceState) {
  24.         super.onCreate(savedInstanceState);
  25.         setContentView(R.layout.activity_rxjava);
  26.         method();
  27.     }
  28.  
  29.     /**
  30.      * RxJava Flowable创建被观察者 背压
  31.      */
  32.  
  33.     private void method() {
  34.         Flowable.interval(1, TimeUnit.MILLISECONDS)
  35.                 .onBackpressureBuffer() // 添加背压策略封装好的方法，此处选择Buffer模式，即缓存区大小无限制
  36.                 .observeOn(Schedulers.newThread())
  37.                 .subscribe(new Subscriber<Long>() {
  38.                     @Override
  39.                     public void onSubscribe(Subscription s) {
  40.                         if (null == s) {
  41.                             return;
  42.                         }
  43.                         mSubscription = s;
  44.                         s.request(Long.MAX_VALUE);//观察者设置可接收最多条数的消息
  45.                     }
  46.  
  47.                     @Override
  48.                     public void onNext(Long aLong) {
  49.                         if (null == aLong) {
  50.                             return;
  51.                         }
  52.                         Log.d("TAG", "Flowable创建被观察者 观察者 onNext方法 aLong----:" + aLong);
  53.                         //模拟1秒接收
  54.                         try {
  55.                             Thread.sleep(1000);
  56.                         } catch (InterruptedException e) {
  57.                             e.printStackTrace();
  58.                         }
  59.                     }
  60.  
  61.                     @Override
  62.                     public void onError(Throwable t) {
  63.                         if (null == t) {
  64.                             return;
  65.                         }
  66.                         Log.d("TAG", "Flowable创建被观察者 观察者 onError方法 t.toString()----:" + t.toString());
  67.                     }
  68.  
  69.                     @Override
  70.                     public void onComplete() {
  71.                         Log.d("TAG", "Flowable创建被观察者 观察者 onComplete方法 ");
  72.                     }
  73.                 });
  74.     }
  75.  
  76.     /**
  77.      * onDestroy方法 取消订阅
  78.      */
  79.  
  80.     @Override
  81.     protected void onDestroy() {
  82.         super.onDestroy();
  83.         if (null != mSubscription) {
  84.             mSubscription.cancel();
  85.             Log.d("TAG", "退出页面 取消订阅");
  86.         }
  87.     }
  88.  
  89. }

 

 

<3> 结果

  1. D/TAG: Flowable创建被观察者 观察者 onNext方法 aLong----:0
  2.  
  3. D/TAG: Flowable创建被观察者 观察者 onNext方法 aLong----:1
  4.  
  5. D/TAG: Flowable创建被观察者 观察者 onNext方法 aLong----:2
  6.  
  7. D/TAG: Flowable创建被观察者 观察者 onNext方法 aLong----:3
  8.  
  9. D/TAG: Flowable创建被观察者 观察者 onNext方法 aLong----:4
  10.  
  11. D/TAG: Flowable创建被观察者 观察者 onNext方法 aLong----:5

 

 

<4> 关闭页面

D/TAG: 退出页面 取消订阅

 

 

<5> 说明

上述代码中有这样的设置

.onBackpressureBuffer() // 添加背压策略封装好的方法，此处选择Buffer模式，即缓存区大小无限制

那么还有没有其他的设置呢，以及每个设置对应的含义是什么呢。

 

举例含义对create操作符
.onBackpressureBuffer()缓存区满时 将缓存区设置无限大 注意内存溢出BUFFER
.onBackpressureDrop()缓存区满时 事件丢失 保留[0,127]条数据 即 前 128条数据DROP
onBackpressureLatest()缓存区满时 事件丢失 保留[0,127]条数据+超出的最后一条数据

LATEST

 

 

 

 

 

 

 

 

 

 

 

四.总结

<1> RxJava的背压可以使用Flowable创建被观察者来实现。背压的使用大前提是 异步订阅中，观察者和被观察者发送消息的速度或者数量不匹配，一般是被观察者发送的速度太快或者消息太多，观察者来不及处理。

 

<2> 背压一般应用在被观察者和观察者异步订阅时，因为同步订阅正式项目很少使用。(RxJava核心就是异步嘛)。

 

<3> 其实一般使用RxJava实现异步，只要注意在适当的时机取消订阅，防止内存泄漏即可。没必要使用背压。

 

<4> 其实处理被观察者和观察者匹配问题也可以有其他的操作办法，比如使用delay()操作符创建一个延时的被观察者。再比如使用interval操作符创建指定时间的被观察者等等。

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/我家自动化/article/detail/300792
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号