在项目中一直都没有怎么用到过这是一大遗憾，只能在demo中使用了。
先给一个RxJAVA3的github 看看里面使用方法，详细就列出来了，可以看看里面解释：
Rxjava3
Rxlifecyc
Retrofit
这里贴上版本信息：
Version 3.x (Javadoc)
single dependency: Reactive-Streams
Java 8+ (Android desugar friendly)
Java 8 lambda-friendly API
fixed API mistakes and many limits of RxJava 2
intended to be a replacement for RxJava 2 with relatively few binary incompatible changes
non-opinionated about the source of concurrency (threads, pools, event loops, fibers, actors, etc.)
async or synchronous execution
virtual time and schedulers for parameterized concurrency
test and diagnostic support via test schedulers, test consumers and plugin hooks
Learn more about RxJava in general on the Wiki Home.Version 2.x
The 2.x version will be supported with bugfixes and important documentation updates until December 31, 2020. No new features will be added to 2.x.Version 1.x
The 1.x version is end-of-life as of March 31, 2018. No further development, support, maintenance, PRs and updates will happen. The Javadoc of the very last version, 1.3.8, will remain accessible.翻译：
版本3.x（Javadoc）
单一依赖项：反应流
Java 8+（Android desugar友好型）
Java 8 lambda友好型API
修正了RxJava 2的API错误和许多限制
打算用相对较少的二进制不兼容更改替换RxJava 2
对竞争源（线程、池、事件循环、光纤、参与者等）不持任何意见
异步或同步执行
参数化竞争的虚拟时间与调度
通过测试调度程序、测试消费者和插件挂钩提供测试和诊断支持
在Wiki主页上了解有关RxJava的更多信息。 版本2.x
在2020年12月31日之前，2.x版本将支持错误修复和重要文档更新。不会在2.x中添加新功能。 版本1.x
截至2018年3月31日，1.x版本已终止。不会进行进一步的开发、支持、维护、减贫战略和更新。最新版本1.3.8的Javadoc将保持可访问性。2.0在今年完就不会更新了。
不啰嗦了，直接开始吧：
package com.lt.rxjavatext.netimport com.lt.rxjavatext.BuildConfig
import okhttp3.OkHttpClient
import okhttp3.Request
import okhttp3.logging.HttpLoggingInterceptor
import retrofit2.Retrofit
import retrofit2.adapter.rxjava2.RxJava2CallAdapterFactory
import retrofit2.converter.gson.GsonConverterFactory
import java.util.concurrent.TimeUnitclass Api private constructor() {companion object {private const val BASE_URL = "https://www.wanandroid.com"const val USER_URL = "/project/tree/json"var retrofit: Retrofit? = nullvar request: ApiService? = nullval mApi: Api by lazy(mode = LazyThreadSafetyMode.SYNCHRONIZED) {Api()}}val log = HttpLoggingInterceptor()fun init() {log.level = if (BuildConfig.DEBUG) HttpLoggingInterceptor.Level.BASIC else HttpLoggingInterceptor.Level.NONE//初始化okhttpval client = OkHttpClient.Builder().connectTimeout(30, TimeUnit.SECONDS)//连接超时设置.writeTimeout(30, TimeUnit.SECONDS)//写入超时设置，.readTimeout(30, TimeUnit.SECONDS)//读取超时设置.addInterceptor(log).build()//初始化retrofitretrofit = Retrofit.Builder().client(client).baseUrl(BASE_URL).addCallAdapterFactory(RxJava2CallAdapterFactory.create()).addConverterFactory(GsonConverterFactory.create()).build()}fun getRequest() {if (request == null) {synchronized(Request::class) {request = retrofit?.create(ApiService::class.java)}}}}上面的地址用到是鸿洋玩安卓里面的api可以用。
仔细的看话就有疑问了，为啥没有用RXJava3 的包，其实已经试验过，直接用Rxjava3是可以用的，但是因为这个方法出问题
.addCallAdapterFactory(RxJava2CallAdapterFactory.create())
忘了把集成贴上了：
implementation "io.reactivex.rxjava3:rxjava:3.0.0-RC2"implementation 'com.squareup.retrofit2:retrofit:2.7.0'implementation 'com.squareup.retrofit2:converter-gson:2.7.0'implementation 'io.reactivex:rxandroid:1.2.1'implementation 'com.squareup.okhttp3:logging-interceptor:3.13.1' //非必要依赖， log依赖，如果需要打印OkHttpLog需要添加implementation 'com.squareup.retrofit2:adapter-rxjava2:2.3.0' // 必要依赖，和Rxjava结合必须用到implementation 'com.trello.rxlifecycle3:rxlifecycle:3.1.0'implementation 'com.trello.rxlifecycle3:rxlifecycle-android-lifecycle-kotlin:3.1.0'implementation 'com.trello.rxlifecycle3:rxlifecycle-components:3.1.0'rxjava3里面包含的有rxjava2里面的基本功能了。但是retrofit没有更新到3，现在里面基本都是android x包了，所有会报错。
在建立一个retrofit2封装的get/post请求 类 class
package com.lt.rxjavatext.netimport com.lt.rxjavatext.bean.UserBean
import com.lt.rxjavatext.bean.WanBean
import io.reactivex.Observable
import retrofit2.http.GET
import retrofit2.http.Urlinterface ApiService {@GETfun listUsers(@Url url: String) : Observable<Response<UserBean>>@GETfun wanList(@Url url: String) : Observable<Response<List<WanBean>>>
}里面的UserBean是创建的类，根据需求创建，Response是个公共封装类
package com.lt.rxjavatext.netdata class Response<T>(var errorCode: Int?, var data: T?, var errorMsg: String?)里面很简单，根据后台规则来
然后再定义一个获取数据的基类：
package com.lt.rxjavatext.netimport android.util.Log
import io.reactivex.Observer
import io.reactivex.disposables.Disposableabstract class BaseObserver<T> : Observer<Response<T>> {val TAG = "BaseObserber"override fun onSubscribe(d: Disposable) {Log.e(TAG, ",,,,,,,,,,,,," + d?.isDisposed)}override fun onNext(value: Response<T>) {if (value.errorCode == 0) {onSuccess(value.data)} else {onFailure(null, value.errorMsg)}}override fun onError(error: Throwable) {Log.e(TAG, ",,,,,,,,,,,,," + error.message)}override fun onComplete() {Log.e(TAG, ",,,,,onComplete,,,,,,,,")}abstract fun onFailure(t: T?, message: String?)abstract fun onSuccess(data: T?)
}这些都是很简单的封装，可以根据自己的需求添加更多的方法
这里需要封装一个调度器，不然很有的重复的代码：
package com.lt.rxjavatext.netimport androidx.lifecycle.LifecycleOwner
import com.trello.rxlifecycle3.android.lifecycle.kotlin.bindToLifecycle
import io.reactivex.ObservableTransformer
import io.reactivex.android.schedulers.AndroidSchedulers
import io.reactivex.schedulers.Schedulers
import java.util.concurrent.TimeUnitobject NetScheduler {fun <T> compose(lifecycleOwner: LifecycleOwner): ObservableTransformer<T, T> {return ObservableTransformer { observable ->observable.subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread()).bindToLifecycle(lifecycleOwner)//绑定生命周期.debounce(1, TimeUnit.SECONDS)//防止1s内重复请求}}
}其实挺遗憾的，除了这里用到了RxLifecycle3,其他的好像没有用到。后续会补上
别忘了初始化再appction里面
class MyAppction :Application(){override fun onCreate() {super.onCreate()Api.mApi.init()Api.mApi.getRequest()}}调用方法
Api.request?.wanList(Api.USER_URL)?.compose(NetScheduler.compose(this))?.subscribe(object :BaseObserver<List<WanBean>>(){override fun onFailure(t: List<WanBean>?, message: String?) {}override fun onSuccess(data: List<WanBean>?) {Log.e("msg","..............${data?.get(0)?.name}.....")}})到这里基本结束了