RxJava基础

╰半橙微兮° 2022-06-01 02:51 173阅读 0赞

**Rx是响应式编程的意思，本质是观察者模式，以观察者（observe）和订阅者（subscribe）为基础的异步响应式开发。观察者模式(Observer)完美的将观察者和被观察的对象分离开。举个例子，用户界面可以作为一个观察者，业务数据是被观察者，用户界面观察业务数据的变化，发现数据变化后，就显示在界面上。RxJava最重要的两个名词是Observables（被观察者，事件源）和Subscribers（观察者）。Observables发出一系列事件，Subscribers处理这些事件**

> Observable：在观察者模式中称为“被观察者； Observer：观察者模式中的“观察者”，可接收Observable发送的数据；  
> subscribe：订阅，观察者与被观察者，通过subscribe()方法进行订阅； Subscriber：也是一种观察者，在2.0中  
> 它与Observer没什么实质的区别，不同的是  
> Subscriber要与Flowable(也是一种被观察者)联合使用，该部分内容是2.0新增的，后续文章再介绍。Obsesrver用于订阅Observable，而Subscriber用于订阅Flowable  
> **subscribeOn(): 指定Observable(被观察者)所在的线程，或者叫做事件产生的线程。 \* observeOn(): 指定 Observer(观察者)所运行在的线程，或者叫做事件消费的线程。**

*  添加依赖

> compile “io.reactivex.rxjava2:rxjava:2.1.1”

**注意：如果需要使用java8的一些新特性，请参考这篇文章配置[点击进入][Link 1]**

*  Hello GJH

public class MainActivity extends AppCompatActivity { 
        private TextView tv;
    
        @Override
        protected void onCreate(Bundle savedInstanceState) {
            super.onCreate(savedInstanceState);
            setContentView(R.layout.activity_main);
            tv= (TextView) findViewById(R.id.tv);
            Observable<String> observable=Observable.create(new ObservableOnSubscribe<String>() {
                @Override
                public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
                    //逻辑处理
                    //。。。
                    //通知观察者
                    e.onNext("Hello GJH");
                }
            });
            Observer<String> observer=new Observer<String>() {
                @Override
                public void onSubscribe(@NonNull Disposable d) {
                    Log.i("info","订阅成功");
                }
    
                @Override
                public void onNext(@NonNull String s) {
                    tv.setText(s);
                    Log.i("info","收到数据"+s);
                }
    
                @Override
                public void onError(@NonNull Throwable e) {
                    Log.i("info","错误发生");
                }
    
                @Override
                public void onComplete() {
                    Log.i("info","完成");
                }
            };
            //绑定
            observable.subscribe(observer);
        }
    }

*  第二种写法（兰姆达表达式）

Observable.just("Hello GJH").subscribe(s->{
                    tv.setText(s);
            });

*  发送集合

//每次发送一条消息 相当于循环调用观察者的onNext
            List<String> list= Arrays.asList("gjh1","gjh2","gjh3","gjh4");
            Observable.fromIterable(list).subscribe(s->{
                //gjh1
                //gjh2
                //gjh3
                //gjh4
                tv.setText(s);
            });

// 发送整数序列 第一个参数代表发送的起始值 第二个参数代表发送的次数(相当于调用了十五次onNext) 发送的值为5 6 7 .。。。19
            Observable.range(5,15).subscribe(s->{
                tv.setText(s.toString());
            });

/*当观察者订阅时，才创建Observable，并且针对每个观察者创建都是一个新的Observable。 以何种方式创建这个Observable对象，当满足回调条件后，就会进行相应的回调 */
    Observable<String> observable = Observable.defer(new Callable<ObservableSource<? extends String>>() {
                @Override
                public ObservableSource<? extends String> call() throws Exception {
                    return Observable.just("hello");
                }
            });

//发送整数序列 每隔三秒发送一次  0 1 2 3 4 .。。。。。。。。。。。。
            Observable.interval(3, TimeUnit.SECONDS).subscribe(s->{
                System.out.print(s);
            });

//延迟两秒发送
            Observable.timer(2, TimeUnit.SECONDS).subscribe(s->{
               });

[Link 1]: http://blog.csdn.net/m0_37402140/article/details/79089048