[移动开发]RxJava简单使用及代码

概述
RxJava 是一个 基于事件流、实现异步操作的库
类似于 Android中的Handler作用

“四剑客”
被观察者（Observable） 产生事件
观察者（Observer） 接收事件并响应
订阅（Subscribe） 关联 被观察者 和 观察者
事件（Event） 被观察者 和 观察者 交互的载体

模型
“被观察者”通过“订阅”按顺序发送“事件”给“观察者”，
“观察者”按顺序接收“事件”作出响应。

用法（代码）
一、依赖

implementation 'io.reactivex.rxjava2:rxandroid:2.1.1'
implementation 'io.reactivex.rxjava2:rxjava:2.2.13'

二、分开的写法
1、创建被观察者以及生产事件

/**
 * Observable.create创造事件
 * ObservableEmitter：事件发射器，通过onNext等方法发送特定事件
 */
val observable: Observable<Int> = Observable.create(object : ObservableOnSubscribe<Int> {
    override fun subscribe(emitter: ObservableEmitter<Int>) {
        emitter.onNext(1)
        emitter.onNext(2)
        emitter.onNext(3)
        emitter.onComplete()
    }
})
2、创建观察者并定义事件响应
val observer: Observer<Int> = object : Observer<Int> {
  override fun onSubscribe(d: Disposable) {
      //默认最先调用复写 onSubscribe（）
      println("onSubscribe" + d)
  }

  override fun onNext(t: Int) {
      println("onNext" + t)
  }

  override fun onError(e: Throwable) {
      println("onError" + e)
  }

  override fun onComplete() {
      println("onComplete")
  }
}
3、通过订阅关联观察者和被观察者
observable.subscribe(observer)

二、链式的写法（简洁）

 Observable.create(object : ObservableOnSubscribe<Int> {
        override fun subscribe(emitter: ObservableEmitter<Int>) {
            emitter.onNext(1)
            emitter.onNext(2)
            emitter.onNext(3)
            emitter.onComplete()
        }
    }).subscribe(object : Observer<Int> {
        override fun onSubscribe(d: Disposable) {
            TODO("Not yet implemented")
        }

        override fun onNext(t: Int) {
            TODO("Not yet implemented")
        }

        override fun onError(e: Throwable) {
            TODO("Not yet implemented")
        }

        override fun onComplete() {
            TODO("Not yet implemented")
        }
    })

三、让观察者不在接受被观察者的事件，但其实被观察者仍然可以继续发

 // 主要是在观察者中写
val observer: Observer<Int> = object : Observer<Int> {
    private var mDisposable: Disposable? = null  // 这个变量能知道是否断开观察
    override fun onSubscribe(d: Disposable) {
        //默认最先调用复写 onSubscribe（）
        println("onSubscribe" + d)
        mDisposable = d
    }

    override fun onNext(t: Int) {
        println("onNext" + t)
        if (t == 2) {
            mDisposable!!.dispose()  // 假设收到2，就让观察者不在接受被观察者的事件
        }
    }

    override fun onError(e: Throwable) {
        println("onError" + e)
    }

    override fun onComplete() {
        println("onComplete")
    }
}

代码文件
kt文件：

package com.example.wantv

import android.os.Bundle
import com.example.view.base.BaseActivity
import com.example.wantv.databinding.ActivityTestRxJavaBinding
import io.reactivex.Observable
import io.reactivex.ObservableEmitter
import io.reactivex.ObservableOnSubscribe
import io.reactivex.Observer
import io.reactivex.disposables.Disposable


/**
 * Create by smy on 2022/2/25
 * Note：
 */
class TestRxJavaActivity : BaseActivity<ActivityTestRxJavaBinding>() {
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        mBinding.btTest.setOnClickListener { test() }
        mBinding.btTest1.setOnClickListener { test1() }
        mBinding.btTest2.setOnClickListener { test2() }
    }

    private fun test2() {
        // 主要是在观察者中写
        val observer: Observer<Int> = object : Observer<Int> {
            private var mDisposable: Disposable? = null  // 这个变量能知道是否断开观察
            override fun onSubscribe(d: Disposable) {
                //默认最先调用复写 onSubscribe（）
                println("onSubscribe" + d)
                mDisposable = d
            }

            override fun onNext(t: Int) {
                println("onNext" + t)
                if (t == 2) {
                    mDisposable!!.dispose()  // 假设收到2，就让观察者不在接受被观察者的事件
                }
            }

            override fun onError(e: Throwable) {
                println("onError" + e)
            }

            override fun onComplete() {
                println("onComplete")
            }
        }
    }

    private fun test1() {
        // 链式写法
        Observable.create(object : ObservableOnSubscribe<Int> {
            override fun subscribe(emitter: ObservableEmitter<Int>) {
                emitter.onNext(1)
                emitter.onNext(2)
                emitter.onNext(3)
                emitter.onComplete()
            }
        }).subscribe(object : Observer<Int> {
            override fun onSubscribe(d: Disposable) {
                TODO("Not yet implemented")
            }

            override fun onNext(t: Int) {
                TODO("Not yet implemented")
            }

            override fun onError(e: Throwable) {
                TODO("Not yet implemented")
            }

            override fun onComplete() {
                TODO("Not yet implemented")
            }

        })
    }

    private fun test() {
        // step1:创建被观察者以及生产事件
        /**
         * Observable.create创造事件
         * ObservableEmitter：事件发射器，通过onNext等方法发送特定事件
         */
        val observable: Observable<Int> = Observable.create(object : ObservableOnSubscribe<Int> {
            override fun subscribe(emitter: ObservableEmitter<Int>) {
                emitter.onNext(1)
                emitter.onNext(2)
                emitter.onNext(3)
                emitter.onComplete()
            }
        })

        /**
         * 其余创造事件的方法
         */
        // 法1：just(T...)：直接将传入的参数依次发送出来。下面会依次调用emitter.onNext(1到4)最后emitter.onComplete()
        var observable1: Observable<Int> = Observable.just(1, 2, 3, 4)
        // 法2：：from(T[])：自动拆分数组依次发送。这个不知道为啥包找不到，注释掉
//        val numbers = arrayOf(1, 2, 3,4,5)
//        val observable2: Observable<Int> = Observable.from(numbers)


        // step2:创建观察者并定义事件响应
        val observer: Observer<Int> = object : Observer<Int> {
            override fun onSubscribe(d: Disposable) {
                //默认最先调用复写 onSubscribe（）
                println("onSubscribe" + d)
            }

            override fun onNext(t: Int) {
                println("onNext" + t)
            }

            override fun onError(e: Throwable) {
                println("onError" + e)
            }

            override fun onComplete() {
                println("onComplete")
            }
        }
        /**
         * 其余创造创建观察者方法，上面方法是基于Observer<>接口
         */
        // 法1：Subscriber 抽象类,也没找到包，直接不写
        /**
         * Subscriber对 Observer 接口进行了扩展，新增了两个方法
         * 1. onStart()：在还未响应事件前调用，用于做一些初始化工作
         * 2. unsubscribe()：用于取消订阅。在该方法被调用后，观察者将不再接收 & 响应事件
         * 调用该方法前，先使用 isUnsubscribed() 判断状态，
         * 确定被观察者Observable是否还持有观察者Subscriber的引用，如果引用不能及时释放，就会出现内存泄露
         */


        // step3：通过订阅关联观察者和被观察者
        observable.subscribe(observer)
    }

}

xml文件：

<?xml version="1.0" encoding="utf-8"?>
<androidx.constraintlayout.widget.ConstraintLayout xmlns:android="http://schemas.android.com/apk/res/android"
    xmlns:app="http://schemas.android.com/apk/res-auto"
    android:layout_width="match_parent"
    android:layout_height="match_parent">

    <Button
        android:id="@+id/bt_test"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:text="复杂写法测试"
        app:layout_constraintLeft_toLeftOf="parent"
        app:layout_constraintTop_toTopOf="parent" />

    <Button
        android:id="@+id/bt_test1"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:layout_marginLeft="10dp"
        android:text="简单写法测试"
        app:layout_constraintLeft_toRightOf="@+id/bt_test"
        app:layout_constraintTop_toTopOf="parent" />

    <Button
        android:id="@+id/bt_test2"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:layout_marginLeft="10dp"
        android:text="主动切断观察，被观察者仍可继续发，但是观察者不收了"
        app:layout_constraintLeft_toRightOf="@+id/bt_test1"
        app:layout_constraintTop_toTopOf="parent" />

</androidx.constraintlayout.widget.ConstraintLayout>

这里mBinding我提前和用viewBinding把V和M关联了，阅读代码只需要把mBinding.btTest看成xml里面的bt_test按钮就行。