Android app 中这样用flow更方便-巧用flow实现polling kotlincoroutine

背景在app开发过程中，实现polling逻辑也是很常见的。当然在移动端应用使用polling处理会影响应用的性能。比如polling处理增加了网络请求的次数，服务端压力增加。polling处理也消耗了更多的网络流量。但是应用polling的场景还是有的。有时是否选择polling要考虑很多综合的因素，比如我们可以使用长连接替代polling，但是长连接在服务端和客户端的开发成本相对要更高些，如果polling只是实现类似的跟帖等功能，我们完全可以使用polling实现，而不是选择代价更高的长连接方案。下面会分使用flow和不使用flow两种方式实现polling并对比两种方式的优缺点。
不使用flow 我们使用线程处理polling请求，首先我们定义了一个polling thread。

class PollingThread: Thread() { override fun run() { var successBlock : (PollingData)->Unit = { Log.d("PollingThread","successBlock $it") } var failBlock:(Exception)->Unit ={ Log.d("PollingThread","failBlock $it") } while (isInterrupted) { pollingApi.call(successBlock, failBlock) Thread.sleep(5000) } } }

在run方法中实现了polling接口的调用，并且接口的调用在while循环中。这里假设polling的时间间隔是5秒钟，所以这里调用线程的sleep方法暂停线程的执行，5秒后再次调用polling接口。polling接口的调用是异步过程，所以这里设置了两个回调，一个用于接收成功的数据，一个用于接收失败的异常。如果在回调中更新了画面，我们还要考虑如何保证回调在ui线程执行，并且回调中不更新消失的页面元素。

class PollingThread(val lifecycleOwner: LifecycleOwner): Thread() { override fun run() { var successBlock : (PollingData)->Unit = { Handler(Looper.getMainLooper()).post { if(lifecycleOwner.lifecycle.currentState >= Lifecycle.State.RESUMED) { Log.d("PollingThread", "successBlock $it") } } } var failBlock:(Exception)->Unit ={ Handler(Looper.getMainLooper()).post { if(lifecycleOwner.lifecycle.currentState >= Lifecycle.State.RESUMED) { Log.d("PollingThread", "failBlock $it") } } } while (isInterrupted) { pollingApi.call(successBlock, failBlock) Thread.sleep(5000) } } }

这段代码增加了回调的线程切换和ui画面有效判断。使用Handler切换线程到ui线程，lifecycler判断ui画面的有效性。
polling线程已经定义完成，下一步我们还要在适当的时机启动polling线程和停止polling线程。

var pollingThread:PollingThread? = null override fun onResume() { super.onResume() pollingThread = PollingThread().run { start() this } } override fun onPause() { super.onPause() pollingThread?.interrupt() pollingThread = null }

这里定义了一个变量pollingThread用于保存启动的polling线程，我们在onResume方法中启动polling线程，在onPause方法中停止线程。经过这样处理后polling就可以工作了。
使用flow 首先我们需要定义一个polling flow。

private val pollingFlow = flow { while (true) { emit(serverApi.getPollingData()) delay(2000) } }

在flow中使用了while循环实现无限轮训，请求的网络接口被定义成了挂起函数，轮训间隔通过协程的delay方法实现。对比不使用flow的方式，polling flow 有自己的一些优势。①无线轮训控制更加简单，不需要复杂逻辑判断，因为flow 中的轮训逻辑中有挂起函数的调用，当收集polling flow的协程被取消时，挂起函数会抛出取消异常，这样就达到了轮训逻辑控制的目的了。②由于调用服务器的接口函数是挂起函数，所以这里避免了使用callback 方法。
我们如何控制线程切换，如何轮训异常呢？

private val pollingFlow = flow { while (true) { emit(serverApi.getPollingData()) delay(5000) } }.flowOn(Dispatchers.IO).retryWhen { cause, attempt -> Log.d("polling flow ", "retryWhen cause $cause attempt $attempt") delay(5000) true }.onEach { Log.d("polling flow ", "onEach $it") } lifecycleScope.launchWhenResumed { pollingFlow.collect() }

我们可以通过flowOn方法切换线程，保证了轮训执行的线程在io线程。在polling flow收集的时候使用默认的ui线程。这样保证了flowOn方法前的部分执行在io线程，flowOn方法后的部分执行在ui线程，进而达到线程切换的目的。这里使用retryWhen方法处理轮训异常，当有异常发生时，延时polling时间间隔后进行重试。
【Android app 中这样用flow更方便-巧用flow实现polling】我们调用了lifecycleScope.launchWhenResumed方法收集flow，这样保证了polling flow只在画面被唤醒的状态下被收集。launchWhenResumed方法是通过切断消息分发来达到挂起的目的，如果在launchWhenResumed方法中又启动了协程进行轮训操作，那么阻止消息分发并不能停止launchWhenResumed方法内部启动协程的轮训操作。在lifecycle-runtime-ktx 2.4.0版本中引入了lifecycle.repeatOnLifecycle方法，这个方法可以根据生命周期进行取消和重启。由于它实现的是协程取消和协程重启，所以在这个方法内部启动的协程也会被取消和重启，进而解决了画面挂起时子协程不被取消而引起的泄露问题。
总结 flow可以充分利用协程的结构化异步的优势实现异步轮训，避免使用启动线程方式进行轮训操作。
线程轮训的方式中延时操作阻塞了线程，flow中的延时操作挂起协程但不阻塞线程，所以flow节省了线程资源，协程挂起时线程还可以处理其他的任务。
创建线程的代价比启动协程的代价更高，并且线程的管理更加麻烦，我们要时刻关心线程状态，控制线程的启动与停止。但是协程依仗结构化异步的特点，用户不需要投入过多的经历管理协程的启动和停止。
使用flow可以通过声明的方式定义polling处理流程，代码逻辑简单清晰。比如通过flow的retryWhen声明重试处理，通过catch捕获polling异常，通过flowOn方法进行线程切换等。
我的公众号已经开通，公众号会同步发布。
欢迎关注我的公众号