[移动开发]Android—Kotlin协程Flow综合应用

前言

在前几篇中，已经讲解了Flow相关的基础知识。在本篇中，将会开启几个小实战来巩固之前所讲解的知识点。

因此阅读本篇所需要的知识点：

• kotlin相关知识点
• jetpack相关知识点

1、准备工作

1.1 先来看看页面整体结构

如图所示

这里准备了五个小案例来进行对应的讲解！

1.2 引入相关的包

    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.4.2'
    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-android:1.4.2'
    implementation "androidx.activity:activity-ktx:1.1.0"
    implementation "androidx.fragment:fragment-ktx:1.2.5"
    def room_version = "2.3.0"
    implementation "androidx.room:room-runtime:$room_version"
    implementation "androidx.room:room-ktx:$room_version"
    kapt "androidx.room:room-compiler:$room_version"
    def nav_version = "2.3.2"
    implementation "androidx.navigation:navigation-fragment-ktx:$nav_version"
    implementation "androidx.navigation:navigation-ui-ktx:$nav_version"
    implementation "androidx.lifecycle:lifecycle-livedata-ktx:2.2.0"
    implementation "androidx.swiperefreshlayout:swiperefreshlayout:1.1.0"
    implementation 'androidx.legacy:legacy-support-v4:1.0.0'
    implementation "com.squareup.retrofit2:retrofit:2.9.0"
    implementation "com.squareup.retrofit2:converter-gson:2.9.0"

1.3 开启ViewBinding

    buildFeatures {
        viewBinding = true
    }

1.4 配置网络权限以及允许http不可少

    <uses-permission android:name="android.permission.INTERNET" />

    <application
        android:allowBackup="true"
        android:icon="@mipmap/ic_launcher"
        android:label="@string/app_name"
        android:networkSecurityConfig="@xml/network_security_config"
        android:roundIcon="@mipmap/ic_launcher_round"
        android:supportsRtl="true"
        android:theme="@style/Theme.FlowPractice">
        <activity android:name=".activity.MainActivity">
            <intent-filter>
                <action android:name="android.intent.action.MAIN" />

                <category android:name="android.intent.category.LAUNCHER" />
            </intent-filter>
        </activity>
    </application>

1.5 network_security_config.xml

<?xml version="1.0" encoding="utf-8"?>
<network-security-config>
    <base-config cleartextTrafficPermitted="true" />
</network-security-config>

准备工作做完了，接下来开始实战！

2、Flow与文件下载

如图所示

这里可以看到，Flow在后台进程里面从服务器里下载数据，然后通过emit发送给对应通道，主线程就通过collect接收对应的数据。

2.1 InputStream 扩展函数

inline fun InputStream.copyTo(out: OutputStream, bufferSize: Int = DEFAULT_BUFFER_SIZE, progress: (Long)-> Unit): Long {
    var bytesCopied: Long = 0
    val buffer = ByteArray(bufferSize)
    var bytes = read(buffer)
    while (bytes >= 0) {
        out.write(buffer, 0, bytes)
        bytesCopied += bytes
        bytes = read(buffer)
        progress(bytesCopied) //在最后调用内联函数
    }
    return bytesCopied
}

这里我们看到，给全局系统类InputStream，额外扩展了copyTo函数，并实现了对应的逻辑。

现在我们来分析一下copyTo函数

• out: OutputStream 这个不用多说，下载保存的对象流

• bufferSize: Int = DEFAULT_BUFFER_SIZE，参数值默认为：DEFAULT_BUFFER_SIZE，可不传

• progress: (Long)-> Unit 内联函数，参数为Long，返回值为Null

  • 也就是说，该方法最后一个参的业务逻辑需要在外部调用时实现！

因此，来看看文件下载的具体逻辑！

2.2 文件下载 DownloadManager

object DownloadManager {
    /**
     * 文件下载
     * @url 下载路径
     * @file 本地保存文件
     */
    fun download(url: String, file: File): Flow<DownloadStatus> {

        return flow {
            val request = Request.Builder().url(url).get().build()
            val response = OkHttpClient.Builder().build().newCall(request).execute()
            if (response.isSuccessful) {
                response.body()!!.let { body ->
                    val total = body.contentLength()
                    //文件读写
                    file.outputStream().use { output ->
                        val input = body.byteStream()
                        var emittedProgress = 0L
                        //使用对应的扩展函数 ，因为该函数最后参为内联函数，因此需要在后面实现对应业务逻辑
                        input.copyTo(output) { bytesCopied ->
                        	//获取下载进度百分比
                            val progress = bytesCopied * 100 / total
                            //每下载进度比上次大于5时，通知UI线程
                            if (progress - emittedProgress > 5) {
                                delay(100)
                                //使用Flow对应的emit 发送对应下载进度通知
                                emit(DownloadStatus.Progress(progress.toInt()))
                                //记录当前下载进度
                                emittedProgress = progress
                            }
                        }
                    }
                }
                //发送下载完成通知
                emit(DownloadStatus.Done(file))
            } else {
                throw IOException(response.toString())
            }
        }.catch {
        	//下载失败，删除该文件，并发送失败通知
            file.delete()
            emit(DownloadStatus.Error(it))
        }.flowOn(Dispatchers.IO) //因为下载文件是属于异步IO操作，因此这里改变上下文

    }

}

这里说明全在注释里，就不过多解释了。不过这里使用到密封类DownloadStatus，来看看具体长啥样：

2.3 DownloadStatus密封类

sealed class DownloadStatus {
    object None : DownloadStatus() //空状态
    data class Progress(val value: Int) : DownloadStatus() //下载进度
    data class Error(val throwable: Throwable) : DownloadStatus() //错误
    data class Done(val file: File) : DownloadStatus() //完成
}

文件下载都准备好了，那看看如何调用！

2.3 使用文件下载

class DownloadFragment : Fragment() {

    val URL = "http://10.0.0.130:8080/kotlinstudyserver/pic.JPG"

	//初始化ViewBinding代码块固定代码
    private val mBinding: FragmentDownloadBinding by lazy {
        FragmentDownloadBinding.inflate(layoutInflater)
    }

    override fun onCreateView(
        inflater: LayoutInflater, container: ViewGroup?,
        savedInstanceState: Bundle?
    ): View? {
    	//布局使用ViewBinding对应的root
        return mBinding.root
    }

    override fun onActivityCreated(savedInstanceState: Bundle?) {
        super.onActivityCreated(savedInstanceState)

        lifecycleScope.launchWhenCreated {
            context?.apply {
                val file = File(getExternalFilesDir(null)?.path, "pic.JPG")
                DownloadManager.download(URL, file).collect { status ->
                    when (status) {
                        is DownloadStatus.Progress -> {
                            mBinding.apply {
                                progressBar.progress = status.value
                                tvProgress.text = "${status.value}%"
                            }
                        }
                        is DownloadStatus.Error -> {
                            Toast.makeText(context, "下载错误", Toast.LENGTH_SHORT).show()
                        }
                        is DownloadStatus.Done -> {
                            mBinding.apply {
                                progressBar.progress = 100
                                tvProgress.text = "100%"
                            }
                            Toast.makeText(context, "下载完成", Toast.LENGTH_SHORT).show()
                        }
                        else -> {
                            Log.d("ning", "下载失败.")
                        }
                    }
                }
            }
        }
    }

}

代码分析：

• lifecycleScope.launchWhenCreated这个表示当控制此 LifecycleCoroutineScope 的 Lifecycle 至少处于 Lifecycle.State.CREATED 状态时，启动并运行给定的块
• 因为 DownloadManager.download(URL, file) 方法里，是使用的 Flow-emit发射值，因此外部需要collect接收对应的值。里面的业务逻辑就是不同状态下的不同处理

来看看运行效果

OK！完美运行，下一个！

3、Flow与Room应用

3.1 先来看看对应的布局结构

如图所示

下面列表就是对应数据库表里面的列表数据，上面按钮表示将输入框的内容添加至数据库对应的用户表里。

3.2 数据表实体类User

@Entity
data class User(
    @PrimaryKey val uid: Int,
    @ColumnInfo(name = "first_name") val firstName: String,
    @ColumnInfo(name = "last_name") val lastName: String
)

这个没啥可说的，下一个。

3.3 对应RoomDatabase

@Database(entities = [User::class], version = 1, exportSchema = false)
abstract class AppDatabase : RoomDatabase() {
    abstract fun userDao(): UserDao

    companion object {

        private var instance: AppDatabase? = null

        fun getInstance(context: Context): AppDatabase {
            return instance ?: synchronized(this) {
                Room.databaseBuilder(context, AppDatabase::class.java, "flow_practice.db")
                    .build().also { instance = it }
            }
        }

    }
}

这个在jetpack专栏里面的详细讲解过，就不再次讲解了。不知道Room使用的：点我查看Jetpack-room讲解

3.4 对应UserDao

@Dao
interface UserDao {

	//返回插入行 ID 的Insert DAO 方法永远不会返回 -1，因为即使存在冲突，此策略也将始终插入行
    @Insert(onConflict = OnConflictStrategy.REPLACE)
    suspend fun insert(user: User)

    @Query("SELECT * FROM user")
    fun getAll(): Flow<List<User>>

}

这里注意的是：在查询所有时，方法getAll的返回值为Flow<List<User>>，最外层为Flow包装！

很简单，快速过一下。

3.5 对应的ViewModel

class UserViewModel(app: Application) : AndroidViewModel(app) {

    fun insert(uid: String, firstName: String, lastName: String) {
        viewModelScope.launch {
            AppDatabase.getInstance(getApplication())
                .userDao()
                .insert(User(uid.toInt(), firstName, lastName))
            Log.d("hqk", "insert user:$uid")
        }
    }

    fun getAll(): Flow<List<User>> {
        return AppDatabase.getInstance(getApplication())
            .userDao()
            .getAll()
            .catch { e -> e.printStackTrace() }
            .flowOn(Dispatchers.IO) //切换上下文为IO异步
    }
}

3.6 对应UI操作

class UserFragment : Fragment() {

    private val viewModel by viewModels<UserViewModel>()
	
	//viewBinding固定代码
    private val mBinding: FragmentUserBinding by lazy {
        FragmentUserBinding.inflate(layoutInflater)
    }
		
    override fun onCreateView(
        inflater: LayoutInflater, container: ViewGroup?,
        savedInstanceState: Bundle?
    ): View? {
        return mBinding.root
    }

    override fun onActivityCreated(savedInstanceState: Bundle?) {
        super.onActivityCreated(savedInstanceState)
        mBinding.apply {
            btnAddUser.setOnClickListener {
                viewModel.insert(
                    etUserId.text.toString(),
                    etFirstName.text.toString(),
                    etLastName.text.toString()
                )
            }
        }
        
        context?.let {
            val adapter = UserAdapter(it)
            mBinding.recyclerView.adapter = adapter
            lifecycleScope.launchWhenCreated {
            	//通过collect 接收 getAll flow发射的数据
                viewModel.getAll().collect { value ->
                    adapter.setData(value)
                }
            }
        }
    }
}

来看看运行效果

完美的运行，下一个。

4、Flow与Retrofit应用

在开始之前，我们来分析一下，Flow与Retrofit之间的关系：

如图所示

• 我们看到用户实时在editText输入的内容通过Flow发送给ViewModel里面的Retrofit，然后Retrofit去请求服务器的数据。
• 服务器下来的数据又通过另一个Flow逐步发射给ViewModel里面的LiveData
• 最后ViewModel的LiveData又实时刷新UI，显示对应的文章内容

服务端代码

public class ArticleServlet extends HttpServlet {

    List<String> data = new ArrayList<>();

    @Override
    public void init() throws ServletException {
        data.add("Refactored versions of the Android APIs that are not bundled with the operating system.");
        data.add("Jetpack Compose is a modern toolkit for building native Android UI. Jetpack Compose simplifies and accelerates UI development on Android with less code, powerful tools, and intuitive Kotlin APIs.");
        data.add("Includes APIs for testing your Android app, including Espresso, JUnit Runner, JUnit4 rules, and UI Automator.");
        data.add("Includes ConstraintLayout and related APIs for building constraint-based layouts.");
        data.add("Includes APIs to help you write declarative layouts and minimize the glue code necessary to bind your application logic and layouts.");
        data.add("Provides APIs for building Android Automotive apps.");
        data.add("A library for building Android Auto apps. This library is currently in beta. You can design, develop, and test navigation, parking, and charging apps for Android Auto, but you can't distribute these apps through the Google Play Store yet. We will make announcements in the future when you can distribute these apps through the Google Play Store.");
        data.add("Provides APIs to build apps for wearable devices running Wear OS by Google.");
        data.add("Material Components for Android (MDC-Android) help developers execute Material Design to build beautiful and functional Android apps.");
        data.add("The Android NDK is a toolset that lets you implement parts of your app in native code, using languages such as C and C++.");
        data.add("The Android Gradle Plugin (AGP) is the supported build system for Android applications and includes support for compiling many different types of sources and linking them together into an application that you can run on a physical Android device or an emulator.");
    }

    @Override
    protected void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
        String key = request.getParameter("key");
        if (key != null) {
            System.out.println(key);
        }else{
            key = "Includes";
        }
        System.out.println("doGet");
        PrintWriter out = response.getWriter();
        JsonArray jsonArray = new JsonArray();
        for (int i = 0; i < data.size(); i++) {
            String text = data.get(i);
            if (text.contains(key)) {
                JsonObject jsonObject = new JsonObject();
                jsonObject.addProperty("id", i);
                jsonObject.addProperty("text", text);
                jsonArray.add(jsonObject);
            }
        }
        out.write(jsonArray.toString());
        System.out.println(jsonArray.toString());
        out.close();
    }
    
}

这个方法类似于文字过滤器，会根据客户端的输入自动匹配相应的文字。

客户端代码

4.1 对应RetrofitClient

object RetrofitClient {

    private val instance: Retrofit by lazy {
        Retrofit.Builder()
            .client(OkHttpClient.Builder().build())
            .baseUrl("http://10.0.0.130:8080/kotlinstudyserver/")
            .addConverterFactory(GsonConverterFactory.create())
            .build()
    }

    val articleApi: ArticleApi by lazy {
        instance.create(ArticleApi::class.java)
    }
}

这里看到使用了ArticleApi，那么

4.2 对应ArticleApi

interface ArticleApi {

    @GET("article")
    suspend fun searchArticles(
        @Query("key") key: String
    ): List<Article>

}

这里就是一个挂起函数，通过key获取网络数据，这里看到数据类Article

4.3 对应Article

data class Article(val id: Int, val text: String)

狠简单，没啥说的。来看看比较核心的ViewModel

4.4 对应ViewModel

class ArticleViewModel(app: Application) : AndroidViewModel(app) {
	
	//定义对应LiveData数据类型
    val articles = MutableLiveData<List<Article>>()

	
    fun searchArticles(key: String) {
        viewModelScope.launch {
            flow {
            	//这里就是通过Retrofit从服务器拿到对应key过滤后的文章内容
                val list = RetrofitClient.articleApi.searchArticles(key)
                //将对应数据发射出去
                emit(list)
            }.flowOn(Dispatchers.IO)
                .catch { e -> e.printStackTrace() }
                .collect {
                	//这里收到对应数据，更新对应的LiveData数据
                    articles.setValue(it)
                }
        }
    }

}

如图所示

对应ViewModel的Flow相当于该图所指向的那部分，其他的就看注释。

现在所有的都准备好了，来看一看对应界面怎么使用的！

4.5 最终调用

class ArticleFragment : Fragment() {
    private val viewModel by viewModels<ArticleViewModel>()

    private val mBinding: FragmentArticleBinding by lazy {
        FragmentArticleBinding.inflate(layoutInflater)
    }

    override fun onCreateView(
        inflater: LayoutInflater, container: ViewGroup?,
        savedInstanceState: Bundle?
    ): View? {
        return mBinding.root
    }

    //获取关键字  分析1 
    private fun TextView.textWatcherFlow(): Flow<String> = callbackFlow {
        val textWatcher = object : TextWatcher {
            override fun beforeTextChanged(s: CharSequence?, start: Int, count: Int, after: Int) {}
            override fun onTextChanged(s: CharSequence?, start: Int, before: Int, count: Int) {}
            override fun afterTextChanged(s: Editable?) {
                offer(s.toString())
            }
        }
        addTextChangedListener(textWatcher)
        awaitClose { removeTextChangedListener(textWatcher) }
    }

    override fun onActivityCreated(savedInstanceState: Bundle?) {
        super.onActivityCreated(savedInstanceState)
        lifecycleScope.launchWhenCreated {
        	//分析2
            mBinding.etSearch.textWatcherFlow().collect {
                Log.d("ning", "collect keywords: $it")
                viewModel.searchArticles(it)
            }
        }

        context?.let {
            val adapter = ArticleAdapter(it)
            mBinding.recyclerView.adapter = adapter
            //分析3
            viewModel.articles.observe(viewLifecycleOwner, { articles ->
                adapter.setData(articles)
            })
        }

    }
}

代码解析

• 分析1：首先这里定义了TextView的扩展函数textWatcherFlow()，因为是private，使其当前所有的TextView都具有textWatcherFlow()方法，而且返回值为Flow<String>

  • 而EditText 最终继承为TextView，因此对应输入框也具备使用对应扩展函数功能

• 分析2：通过分析1可知道，当前页面所有TextView都具有textWatcherFlow()方法，因此可以通过collect来监听文本改变时的内容，对应it就为当前改变的值

• 分析3：因为viewModel.articles是对应ViewModel里面的LiveData数据，因此通过.observe实时监听LiveData的改变，并且实时刷新对应的adapter

来看看运行效果

对应后台打印日志

OK！完美的运行！下一个！

5、冷流还是热流

• Flow是冷流，那什么是冷流？简单来说，如果Flow有了订阅者Collector以后，发射出来的值，才会实实在在的存在与内存之中，这根懒加载的概念很像。
• 与之相对的是热流，StateFlow和SharedFlow是热流，在垃圾回收之前，都是存在内存之中，并且处于活跃状态！

那么！

5.1 StateFlow

StateFlow是一个状态容器式可观察数据流，可以向其收集器发出当前状态更新和新状态更新。骇客通过其value属性读取当前状态值。

概念说完了，开始实战试试手：

5.1.1 对应ViewModel

class NumberViewModel : ViewModel() {

    val number = MutableStateFlow(0)

    fun increment() {
        number.value++
    }

    fun decrement() {
        number.value--
    }
}

这看着好像LiveData，先不管，接着看对应使用！

5.1.2 具体使用

class NumberFragment : Fragment() {
    private val viewModel by viewModels<NumberViewModel>()

    private val mBinding: FragmentNumberBinding by lazy {
        FragmentNumberBinding.inflate(layoutInflater)
    }

    override fun onCreateView(
        inflater: LayoutInflater, container: ViewGroup?,
        savedInstanceState: Bundle?
    ): View? {
        return mBinding.root
    }

    override fun onActivityCreated(savedInstanceState: Bundle?) {
        super.onActivityCreated(savedInstanceState)
        mBinding.apply {
            btnPlus.setOnClickListener {
                viewModel.increment()
            }

            btnMinus.setOnClickListener {
                viewModel.decrement()
            }
        }

        lifecycleScope.launchWhenCreated {
            viewModel.number.collect { value ->
                mBinding.tvNumber.text = "$value"
            }
        }
    }
}

使用的方式也和LiveData的方式如出一辙，也很简单

来看看运行效果

整体使用感觉和LiveData差不多！那看看SharedFlow怎么使用的？

5.2 SharedFlow

如图所示

该页面有三个Fragment，每个Fragment里面都是只有一个TextView。下方有开始和停止两个按钮。

想要实现三个Fragment，秒表同步走动的效果！

5.2.1 先看主Fragment

class SharedFlowFragment : Fragment() {

    private val viewModel by viewModels<SharedFlowViewModel>()

    private val mBinding: FragmentSharedFlowBinding by lazy {
        FragmentSharedFlowBinding.inflate(layoutInflater)
    }

    override fun onCreateView(
        inflater: LayoutInflater, container: ViewGroup?,
        savedInstanceState: Bundle?
    ): View? {
        return mBinding.root
    }

    override fun onActivityCreated(savedInstanceState: Bundle?) {
        super.onActivityCreated(savedInstanceState)
        mBinding.apply {
            btnStart.setOnClickListener {
                viewModel.startRefresh()
            }

            btnStop.setOnClickListener {
                viewModel.stopRefresh()
            }
        }
    }

}

这里没啥可说的，就两个按钮，分别调用了ViewModel不同的方法！

5.2.2 来看看SharedFlowViewModel

class SharedFlowViewModel : ViewModel() {

    private lateinit var job: Job

    fun startRefresh() {
        job = viewModelScope.launch(Dispatchers.IO) {
            while (true) {
                LocalEventBus.postEvent(Event((System.currentTimeMillis())))
            }
        }
    }

    fun stopRefresh() {
        job.cancel()
    }

}

这里我们可以看到，方法startRefresh里面开启了IO协程，里面使用了LocalEventBus.postEvent。

5.2.3 来看看LocalEventBus

object LocalEventBus {
    val events = MutableSharedFlow<Event>()
    suspend fun postEvent(event: Event) {
        events.emit(event)
    }

}

data class Event(val timestamp: Long)

这里可以看到每调用一次postEvent方法，都会将当前时间通过emit发射出去

既然有发射，那么肯定会有接收端！

5.2.4 次级Fragment

class TextFragment : Fragment() {


    private val mBinding: FragmentTextBinding by lazy {
        FragmentTextBinding.inflate(layoutInflater)
    }

    override fun onCreateView(
        inflater: LayoutInflater, container: ViewGroup?,
        savedInstanceState: Bundle?
    ): View? {
        return mBinding.root
    }

    override fun onActivityCreated(savedInstanceState: Bundle?) {
        super.onActivityCreated(savedInstanceState)
        lifecycleScope.launchWhenCreated {
            LocalEventBus.events.collect {
                mBinding.tvTime.text = it.timestamp.toString()
            }
        }
    }

}

可以看到，直接通过LocalEventBus.events.collect来接收，postEvent发射过来的值，并且实时改变文本内容！

来看看运行效果

完美运行！

结束语

好了，本篇到这里就结束了，通过一系列综合应用，相信读者对Flow以及对应Jetpack有了更加深刻的印象！在下一篇中，将会先讲解Jetpack对应的Paging组件内容！