当Kotlin遇见RxJava多数据源

资讯频道 → 编程语言

0顶
1踩

2017-05-25 10:19 by 副主编 jihong10102006 评论(0) 有6274人浏览

Kotlin RxJava 数据源

声明：ITeye资讯文章的版权属于ITeye网站所有，严禁任何网站转载本文，否则必将追究法律责任！

引用

作者：Code4Android
来源：http://blog.csdn.net/xiehuimx/article/details/72643606

温馨提醒

阅读本文最好有Kotlin基础，若没有基础，可参考之前文章Kotlin初探，使用Kotlin优雅的开发Android应用，以及RxJava基础（本文基于RxJava2），当然我也会尽可能详细解释让你顺利阅读本文。
源码传送门

写在前面

最近几天回过头，看了之前的总结RxJava操作符系列，感觉对Rxjava多数据源的处理不是很理解，所以在总结学习一波。大家都知道，最近Kotlin语言一直占据热搜榜，褒贬不一，但我想说，不管有什么想法都要抛在脑后，毕竟Google爸爸出手，你不情愿也要跟随它的步伐。鉴于此，本篇对RxJava多数据源的总结是基于Kotlin语言，也让大家明白，使用Kotlin开发应用并不是不能使用Java库，现在有一部分人担心，Kotlin第三方库那么少，如果使用Kotlin开发那不是给自己找罪受，其实你完全错了，当你说这话的时候，我敢断定你都还没有接触Kotlin，因为Koltin有一个最重要的优势就是和Java绝对兼容。

多数据源处理操作符

在RxJava中多数据源处理的操作符很多，但是最经典的就要数merge,contact，zip了。如果对这三个操作符不是很熟悉的话，可以去查看它的使用，当然如果你懒得去看，我也会简单提一下。merge操作符可以处理多个Observable发送的数据，它是一个异步操作，不保证数据发送的顺序，即有可能出现数据交叉，当一个Observable发送了onError后，未执行的Observable不在继续执行，直接执行merge的onError方法。

contact操作符执行时一个同步操作，严格按照contact中传入Observable先后执行，即前面的先执行后面的后执行，并且最终发送的数据也是有序的，即第一个Observable的数据发送完毕再发送第二个，依次类推。

zip操作符和contact和merge有了本质的区别，它会将每个Observable个数据项分布对应返回一个Observable再发送，最终发送的数据量与最小数据长度相同。

使用场景分析

假如现在我们有三种商品，有一个查询商品信息的接口，根据接口可以查询该商品的价格以及出售地点。商品实体类

data class Goods(var id:Int,var price: Int, var address: String)

在Kotlin语言中，实体类创建用data class 关键词，我们不需要和Java一样创建get/set方法，只需一行代码搞定。

创建模拟网络请求

object NetRequest {
    //模拟网络请求
    fun getGoodsObservable(id: Int): Observable<Goods> {
fun getGoodsObservable(id: Int): Observable<Goods> {
        return Observable.create {
            source ->
            Thread.sleep(Random().nextInt(1000).toLong())
            var data = Goods(id, Random().nextInt(20), "地址${id}")
            source.onNext(data)
            source.onComplete()
            Log.e("getGoodsObservable:", "${id}")
        }
    }
}

在上面我们创建了一个单例类，在Kotlin中使用object修饰类时即给我们自动创建了一个单例对象。在每一句代码结尾我们不需要再和Java一样写一个分号“；”来结束，什么也不用写。

Observable.create使用的是lambda表达式，在Kotlin语言中是支持lambda表达式的。source 就是ObservableEmitter，所以我们可以调用onNext发送数据。为了更准确的模拟网络请求，使用Thread.sleep随机的延迟，模拟网络请求的时间。

  fun getGoodsObservable(id: Int): Observable<Goods> {
        return Observable.create {
            source ->
            Thread.sleep(Random().nextInt(1000).toLong())
            var data = Goods(id, Random().nextInt(20), "地址${id}")
            source.onNext(data)
            source.onComplete()
            Log.e("getGoodsObservable:", "${id}")
        }

当然由于subscribe只有一个参数，所以我们也可以这样写。也就是省略了source ->，此时it就表示该参数数据。

return Observable.create {
            Thread.sleep(Random().nextInt(1000).toLong())
            var data = Goods(id, Random().nextInt(20), "地址${id}")
            it.onNext(data)
            it.onComplete()
            Log.e("getGoodsObservable:", "${id}")
        }

在java中实现如下

  return Observable.create(new ObservableOnSubscribe<Goods>() {

            @Override
            public void subscribe(@NonNull ObservableEmitter<Goods> e) throws Exception {
             //处理逻辑
            }
        });

merge

准备好了请求操作，开始使用merge看看执行的效果。

  fun executeMerge() {
        Observable.merge(getGoodsObservable(1).subscribeOn(Schedulers.newThread()),
                getGoodsObservable(2).subscribeOn(Schedulers.newThread()),
                getGoodsObservable(3).subscribeOn(Schedulers.newThread()))
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .toList()
                .subscribe({
                    Log.e(TAG, it.toString())
                }, {
                    Log.e(TAG, it.toString())
                })
    }

merge中有三个网络请求操作，并通过subscribeOn(Schedulers.newThread())将网络请求切换到线程中执行，数据都请求成功后，再通过observeOn(AndroidSchedulers.mainThread())切换到主线程请求数据。为了三请求都成功后，我们在更新UI，所以通过toList()将请求的数据转换成List一块发送。在上面的subscribe依然使用的lambda表达式，subscribe（{}，{}）中第一个括号是onSuccess回调，里面的it是接收到的List< Goods >数据，第二个括号是onError回调，it表示异常Throwable对象。
subscribe部分Java代码

.subscribe(new Consumer<List<Goods>>() {
                    @Override
                    public void accept(@NonNull List<Goods> goodses) throws Exception {

                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(@NonNull Throwable throwable) throws Exception {

                    }
                });

当然如果你想使用RxJava2中onSubscribe(@NonNull Disposable d) ，你可以这样使用subscribe

.subscribe(object : SingleObserver<List<Goods>> {
                    override fun onSubscribe(d: Disposable?) {
                    }
                    override fun onError(e: Throwable?) {
                    }
                    override fun onSuccess(t: List<Goods>?) {
                    }
                })

为了观察，我们将请求成功的数据显示在界面上，我们创建一个Button，TextView。

class MainActivity : AppCompatActivity(), View.OnClickListener {

    val TAG = "MainActivity"
    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)
        setSupportActionBar(toolbar)
        //加入这句import kotlinx.android.synthetic.main.activity_main.*
        //不用再findViewById,可直接使用
        merge.setOnClickListener(this)

    }
    override fun onClick(v: View) {
        when (v.id) {
            R.id.merge -> {
                executeMerge()
            }
        }
        //when 关键字和Java中的Switch关键词是类似的，
        //只不过它比Java中的Switch强大的多，可以接收任何参数，
        //然后判断使用,也可以如下使用
        when (v) {
            merge -> {
            }
        }
    }
}

contact

我们点击执行几次发现，返回的List的数据并不是按照merge参数的先后顺序执行的，它是并发的，最终的顺序，是由网络请求的快慢决定的，请求返回数据越快也就表示该数据最早发送，即在List中最靠前。那么此时出现一个问题，如果我想返回数据的List顺序严格按照位置的先后顺序呢？那此时使用merge的话，是不太现实了。当然前面我们提到contact可以使用。那么直接将merge更改为contact执行以下试试，

    fun executeContact() {
        Observable.concat(getGoodsObservable(1).subscribeOn(Schedulers.newThread()),
                getGoodsObservable(2).subscribeOn(Schedulers.newThread()),
                getGoodsObservable(3).subscribeOn(Schedulers.newThread()))
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .toList()
                .subscribe({
                    Log.e(TAG, it.toString())
                }, {
                    Log.e(TAG, it.toString())
                })
    }

的确，发现无论执行多少次List的数据都能按照contact中Observable顺序发送，我们想要的效果可以实现了，不过你会发现，效率太差了，这是同步执行啊，只有第一个请求成功，才会去请求第二个，然后第三个，假如一次请求需要一秒，那三次请求至少三秒啊，不能忍。

zip

鉴于上面两种方式的利弊，如果我们既想如merge一样并发执行，又想和contact一样保证顺序，是不是有点强迫症的意思，当然强大的zip就能实现我们想要的效果。如下实现。

    fun executeZip() {
        Observable.zip(getGoodsObservable(1),
                getGoodsObservable(2),
                getGoodsObservable(3),
                Function3<Goods, Goods, Goods, List<Goods>>
                { goods0, goods1, goods2 ->
                    val list = ArrayList<Goods>()
                    list.add(goods0)
                    list.add(goods1)
                    list.add(goods2)
                    list
                }).subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe({
                    Log.e(TAG, it.toString())
                }, {
                    Log.e(TAG, it.toString())
                })
    }

既然实现了，那我们运行几次，发现完美的实现了我们想要的效果，即并发的执行了，也保证了我们请求数据的顺序性。

在回调中运用RxJava

在上面我们的单个网络请求是一个同步的请求，如果我们的网络请求封装了，在线程中请求，请求成功后在主线程中回调，那我们又该如何创建呢使用呢？
先来模拟一个子线程请求网络，请求成功回调数据给主线程。

    fun getGoods(ctx:Context,id: Int,callbacks:(goods:Goods)->Unit): Unit {
        ctx.doAsync {
            Thread.sleep(Random().nextInt(1000).toLong())
            var data = Goods(id, Random().nextInt(20), "地址${id}")
            ctx.runOnUiThread {
                callbacks(data)
            }
        }
    }

getGoods传了三个参数，第一个Context对象，第二个是商品ID,第三个参数是一个函数，(goods:Goods)->Unit表示第三个参数的类型是一个参数为Goods类型并且返回Unit的函数。使用doAsync 模拟异步请求，请求成功后runOnUiThread 切换到UI线程。然后callbacks（data）将数据回调。这种使用方式比Java中回调优美好用太多了。
接下来就开始在回调成功后创建Observable

  fun getGoodsCallBack(id: Int): Observable<Goods> {
        var subscrbe: ObservableEmitter<Goods>? = null
        var o = Observable.create<Goods> {
            subscrbe = it
        }
        //Kotlin特性
        getGoods(this@MainActivity, id) {
            subscrbe?.onNext(it)
        }
        return o
    }
    fun executeZipCallBack() {
        Observable.zip(getGoodsCallBack(1).subscribeOn(Schedulers.newThread()),
                getGoodsCallBack(2).subscribeOn(Schedulers.newThread()),
                getGoodsCallBack(3).subscribeOn(Schedulers.newThread()),
                Function3<Goods, Goods, Goods, List<Goods>>
                { goods0, goods1, goods2 ->
                    val list = ArrayList<Goods>()
                    list.add(goods0)
                    list.add(goods1)
                    list.add(goods2)
                    list
                }).subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe({
                    Log.e(TAG, it.toString())
                }, {
                    Log.e(TAG, it.toString())
                })
    }

ok,到这里回调情况下创建使用RxJava也介绍完毕，到此本篇文章就结束了，有问题欢迎指出，内容杂乱，多多担待，Hava a wonderful day.

分享到：

0
顶

1
踩

评论共 0 条请登录后发表评论

发表评论

您还没有登录,请您登录后再发表评论

相关推荐

Kotlin

Kotlin是JetBrains团队开发的一门基于 JVM 的，现代的、注重工程实用性的静态类型编程语言。Kotlin可以编译成Java字节码，也可以编译成JavaScript、Native、而且它是由 Jetbrains 开发的。Kotlin 编译为字节码，则...

【新收录】CSDN日报 —— Kotlin 专场

谷歌在 I/O 2017 开发者大区会上宣布了 Kotlin 成为安卓开发一级语言。那么，Kotlin 会取代 Java 么，你怎么看？

CSDN日报20170525 ——《狗蛋的寻亲之路，结识嵌入式世界的那只狗》

当 Kotlin 遇见 RxJava 多数据源作者： Code4Android 本篇对 RxJava 多数据源的总结是基于 Kotlin 语言，也让大家明白，使用 Kotlin 开发应用并不是不能使用 Java 库，现在有一部分人担心，Kotlin 第三方...

2019年为android开发准备的面试题（含答案）

Integer是int的包装类，int则是java的一种基本数据类型，Integer变量必须实例化才能使用，当new一个Integer时，实际是生成一个指向此对象的引用，而int是直接存储数据的值，Integer默认值是null，而int默认值是0 ...

Jetpack Room,Navigation,ViewModel

Jetpack Gayhub地址本页面内容仅为个人学习笔记，受限于微博的能力，可能存在一定概念或者理解上的问题。 [外链图片转存失败,源站可能有...[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-

[转载] 【汇总】Android知识清单

服务器只提供数据接收接口，在多线程或多进程条件下，如何保证数据的有序到达？动态布局的理解怎么去除重复代码？画出 Android 的大体架构图 Recycleview和ListView的区别 ListView图片加载错乱的原理...

Android高级面试全记录（刷题，不断补充完善）

4.ContentProvider 进程(应用程序)间数据共享,数据源可以是sqlite,也可以是xml,相关类: ContentResolver(内容解析器), ContentObserver(数据观察者) 5. 网络存储天气数据的xml,json格式等等,通过...

总结2022最全的BAT大厂面试题整理及分析

熟练掌握java是很关键的，大公司不仅仅要求你会使用几个api，更多的是要你熟悉源码实现原理，甚至要你知道有哪些不足，怎么改进，还有一些java有关的一些算法，设计模式等等。这里整理的是一些与技术没有直接关系的...

应该是史上最全最新Java和Android面试题目(自己总结和收集的)

ConcurrentHashMap线程安全，引入“分段锁”概念，通过将整个map分为多个segment（类似HashTable）可以提供相同的线程安全，效率默认提供16倍 HashTable, HashMap，TreeMap区别？数组和链表的区别 ...

COMSOL模拟碳酸钙岩石与盐酸反应的随机孔隙酸化路径及布林克曼流动形成的分形结构

内容概要：本文详细介绍了利用COMSOL软件模拟碳酸钙(CaCO3)在岩石中与盐酸(HCl)反应过程中产生的随机孔隙酸化路径及其形成的布林克曼流动。首先，通过蒙特卡洛方法生成随机孔隙分布，模拟真实岩石内部复杂的孔隙结构。接着，采用布林克曼方程处理多孔介质中的粘性力和渗透流动，并引入化学反应模块，模拟CaCO3与HCl之间的化学反应。随着模拟的进行，酸液流动路径逐渐形成类似雪花状的分形结构，展示了流动与溶解之间的动态博弈。最后，通过自适应网格技术和粒子追踪功能，精确捕捉并可视化这些精美的分形图案。适合人群：从事地质工程、材料科学、化学工程等领域研究的专业人士，以及对多孔介质传输现象感兴趣的科研工作者。使用场景及目标：适用于研究多孔介质内的化学反应和流体流动特性，特别是对于优化石油开采中的酸化压裂工艺具有重要指导意义。其他说明：文中提供了详细的MATLAB和COMSOL代码片段，帮助读者理解和重现模拟过程。此外，强调了随机性和确定性在微观尺度上的相互作用，揭示了自然界深层次的规律。

基于滑模控制的永磁同步电机直接转矩控制仿真建模与实现

内容概要：本文详细介绍了将滑模控制(SMC)应用于永磁同步电机(PMSM)直接转矩控制(DTC)的技术细节。首先解释了转矩和磁链误差计算方法，接着探讨了滑模面的设计及其对系统抖振的影响。文中还提供了扇区矢量选择的具体实现方式，并深入讨论了磁链观测器的改进措施。此外，文章分析了滑模控制器的设计要点以及仿真过程中需要注意的关键参数配置。通过对比传统PI控制，验证了滑模控制在提高系统鲁棒性和快速响应方面的优势。适合人群：从事电机控制系统研究的专业人士，尤其是对永磁同步电机直接转矩控制感兴趣的科研工作者和技术人员。使用场景及目标：适用于希望深入了解并掌握滑模控制理论及其在PMSM-DTC应用中的具体实现方法的研究人员。目标是在实际项目中能够运用滑模控制提升系统的稳定性和性能。其他说明：文中提供的MATLAB/Simulink代码片段有助于读者更好地理解和复现实验结果。同时提醒读者关注一些常见的陷阱，如参数选择不当可能导致的问题。

北京大学网络安全工作人员管理规定：涵盖人员职责、聘用、转岗离岗、教育培训及第三方管理

内容概要：本文详细介绍了北京大学针对网络安全工作人员的管理规定，旨在加强网络安全管理和明确不同角色的责任。全文分为九章，涵盖了网络安全工作人员及其职责、聘用管理、转岗和离岗管理、教育培训、第三方人员管理及奖惩措施等方面的内容。重点在于明确各级单位和人员的具体职责，确保网络安全制度的有效执行，并强调了对第三方人员的严格管控和保密要求。适合人群：适用于高校网络安全管理人员及相关技术人员，尤其是北京大学及其下属单位的网络安全工作者。使用场景及目标：①帮助高校建立健全网络安全管理体系；②指导网络安全工作人员明确自身职责，提高工作效率；③规范第三方人员的访问和操作，降低安全风险。其他说明：本文还提供了多个附件，如网络安全承诺书、访问申请表和保密协议模板，便于实际操作和管理。

网络设备市场现状与发展趋势分析（2024-2030年）-技术革新与智能化应用

内容概要：本文深入探讨了中国网络设备市场的现状及其未来发展潜力。首先介绍了网络设备的基本概念及其作为现代通信网络基础设施的重要地位，随后分析了当前市场面临的挑战和技术进步带来的机遇。文中特别强调了5G、物联网、云计算等新兴技术对网络设备性能和安全性的更高要求，以及由此催生的高带宽、低延迟产品的市场需求。此外，还讨论了软件定义网络(SDN)、网络功能虚拟化(NFV)、边缘计算等新技术的应用前景，指出未来网络设备将更加智能化、自动化，并能更好地支持AI和ML技术。最后，通过对多家领先企业的案例研究，展示了行业内竞争态势及各公司在技术创新方面的努力。适用人群：从事网络设备相关领域的研究人员、工程师、管理人员，以及关注该领域发展的投资者。使用场景及目标：帮助读者了解网络设备行业的最新动态和技术趋势，为制定战略决策提供依据；同时为企业和个人投资者提供市场洞察，辅助其做出合理的投资选择。其他说明：报告基于详实的数据分析和专家意见撰写而成，旨在为专业人士提供有价值的参考资料。

西门子1200 PLC码垛系统的SCL编程详解：涵盖变频器、机器人、视觉系统集成

内容概要：本文详细介绍了基于西门子1200 PLC的码垛系统的设计与实现，涵盖了多个关键技术点。首先，文章讲解了Modbus TCP通讯的实现方法，展示了如何通过TSEND_C和TRCV_C功能块进行工业相机和机器人之间的数据传输，并提供了具体的报文处理代码。接着，文章深入探讨了SCL编程的优势及其在复杂逻辑处理中的应用，如托盘堆叠算法，该算法能够根据当前层数动态调整机械手的高度，确保堆叠的安全性和稳定性。此外，文章还介绍了机器人控制中的移位寄存器实现的动作队列管理和变频器的速度平滑处理，以及视觉系统的坐标解析和异常处理机制。最后，文章强调了良好的注释规范和异常处理链的重要性，确保程序的可维护性和可靠性。适合人群：从事工业自动化领域的工程师和技术人员，尤其是熟悉西门子PLC编程和SCL语言的从业者。使用场景及目标：适用于需要深入了解和掌握西门子1200 PLC在码垛系统中的具体应用的技术人员。目标是帮助读者理解并实现多设备联动的复杂控制系统，提高系统的稳定性和效率。其他说明：文中提供的代码示例和详细的解释有助于读者更好地理解和应用相关技术，同时也为后续的维护和优化提供了宝贵的参考资料。

ZYNQ平台PS与PL端驱动程序编写

适合从入门到进阶的驱动程序爱好者

计算机二级上机题库答案.pdf

深信服下一代防火墙：构建全方位立体网络安全监测与响应体系

内容概要：本文介绍了深信服科技推出的下一代防火墙（NGAF）网络安全监测解决方案。随着网络安全成为国家战略的一部分，企业不仅需要遵守法律法规，还需增强自身的网络安全防护能力。传统的安全措施难以应对复杂的新型威胁，如APT攻击。深信服的NGAF通过多维度的安全监测，包括入侵风险、僵尸主机、实时漏洞、数据风险、黑链风险以及对外DoS攻击监测等功能，结合云端威胁情报共享，为企业提供了一套立体化的主动防御体系。该方案不仅可以旁路或串接部署，不影响现有业务系统，还能通过外置数据中心进行日志管理和综合分析，帮助用户快速定位和解决安全问题。适合人群：IT管理人员、网络安全专家、企业信息安全负责人。使用场景及目标：适用于各类企业的网络安全建设，特别是需要应对复杂网络攻击的企业。目标是构建一个多层次、全方位、智能化的网络安全监测和响应体系，提高企业的安全防护能力和应急响应速度。其他说明：深信服NGAF不仅提升了网络安全监测的效果，还降低了运维成本，改变了传统的被动防护模式，使得安全运维更加高效和智能化。

COMSOL超表面偏振转换技术：介质半波片与1/4波片的设计与仿真

内容概要：本文详细介绍了利用COMSOL软件进行超表面偏振转换的设计方法，主要聚焦于介质半波片和1/4波片的实现。文中首先解释了超表面的基本原理及其在光学调控中的重要作用，随后具体阐述了如何在COMSOL中设置材料属性、创建几何结构并施加适当的边界条件。针对半波片和1/4波片的不同需求，分别探讨了它们各自的设计要点、模拟步骤及优化策略。此外，还分享了一些实用的编码示例和技术诀窍，帮助研究人员更好地理解和掌握相关技能。适合人群：从事光学工程、光电子学等领域研究的专业人士，尤其是那些希望深入了解超表面偏振转换机制并对COMSOL有一定使用经验的技术人员。使用场景及目标：适用于需要设计高性能偏振转换器件的研究项目，旨在提高对超表面特性的认识水平，推动新型光学组件的研发进程。通过学习本文提供的理论知识和实践经验，读者可以在实际工作中运用COMSOL完成高质量的仿真实验。其他说明：文中不仅涵盖了基本的概念介绍，还包括了许多具体的实施细节，如参数选择、模型构建、边界条件设定等，这些都是成功搭建有效仿真的关键因素。同时，作者也强调了实验过程中可能出现的问题及解决方案，为后续研究提供了宝贵的参考资料。

0顶1踩