Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

Kotlin ทำงานอย่างลื่นไหลใน Android

ในโครูทีน Flow เป็นประเภทที่สามารถปล่อยค่าหลายค่า ตามลำดับ ซึ่งต่างจากฟังก์ชันระงับที่แสดงผลค่าเดียวเท่านั้น เช่น คุณสามารถใช้โฟลว์เพื่อรับข้อมูลอัปเดตแบบเรียลไทม์จากฐานข้อมูลได้

Flow สร้างขึ้นจากโครูทีนและสามารถให้ค่าได้หลายค่า โดยแนวคิดแล้ว Flow คือสตรีมข้อมูลที่สามารถคำนวณแบบอะซิงโครนัสได้ ค่าที่ปล่อยออกมาต้องเป็นประเภทเดียวกัน ตัวอย่างเช่น Flow<Int> คือโฟลว์ที่ปล่อยค่าจำนวนเต็ม

Flow คล้ายกับ Iterator มาก ซึ่งสร้างลำดับของค่า แต่ใช้ฟังก์ชันระงับเพื่อสร้างและใช้ค่าแบบไม่พร้อมกัน ซึ่งหมายความว่า ตัวอย่างเช่น โฟลว์สามารถส่งคำขอเครือข่ายเพื่อสร้างค่าถัดไปได้อย่างปลอดภัยโดยไม่บล็อกเทรดหลัก

เอนทิตีที่เกี่ยวข้องกับสตรีมข้อมูลมี 3 รายการ ได้แก่

ผู้ผลิตสร้างข้อมูลที่จะเพิ่มลงในสตรีม Flow ยังสร้างข้อมูลแบบไม่พร้อมกันได้ด้วย โครูทีน
(ไม่บังคับ) ตัวกลางสามารถแก้ไขค่าแต่ละค่าที่ปล่อยลงใน สตรีมหรือสตรีมเองได้
ผู้ใช้จะใช้ค่าจากสตรีม

เอนทิตีที่เกี่ยวข้องในสตรีมข้อมูล ผู้บริโภค ตัวกลาง และผู้ผลิต (ไม่บังคับ) — **รูปที่ 1** เอนทิตีที่เกี่ยวข้องในสตรีมข้อมูล ได้แก่ ผู้บริโภค ตัวกลาง (ไม่บังคับ) และผู้ผลิต

ใน Android ที่เก็บมักจะเป็นผู้ผลิตข้อมูล UI ที่มีอินเทอร์เฟซผู้ใช้ (UI) เป็นผู้บริโภค ซึ่งจะแสดงข้อมูลในท้ายที่สุด ในบางครั้ง เลเยอร์ UI จะเป็นผู้สร้าง เหตุการณ์อินพุตของผู้ใช้ และเลเยอร์อื่นๆ ในลําดับชั้นจะใช้เหตุการณ์เหล่านั้น เลเยอร์ระหว่างผู้ผลิตและผู้บริโภคมักจะทำหน้าที่เป็นตัวกลางที่แก้ไขสตรีมข้อมูลเพื่อปรับให้ตรงกับข้อกำหนดของเลเยอร์ถัดไป

การสร้างโฟลว์

หากต้องการสร้างโฟลว์ ให้ใช้ API ของ เครื่องมือสร้างโฟลว์ ฟังก์ชันตัวสร้าง flow จะสร้างโฟลว์ใหม่ที่คุณสามารถส่งค่าใหม่ลงในสตรีมข้อมูลได้ด้วยตนเองโดยใช้ฟังก์ชัน emit

ในตัวอย่างต่อไปนี้ แหล่งข้อมูลจะดึงข่าวสารล่าสุดโดยอัตโนมัติในช่วงเวลาที่กำหนด เนื่องจากฟังก์ชันระงับไม่สามารถ แสดงผลค่าที่ต่อเนื่องกันหลายค่า แหล่งข้อมูลจึงสร้างและแสดงผล โฟลว์เพื่อให้เป็นไปตามข้อกำหนดนี้ ในกรณีนี้ แหล่งข้อมูลจะทำหน้าที่เป็นผู้ผลิต

class NewsRemoteDataSource(
    private val newsApi: NewsApi,
    private val refreshIntervalMs: Long = 5000
) {
    val latestNews: Flow<List<ArticleHeadline>> = flow {
        while (true) {
            val latestNews = newsApi.fetchLatestNews()
            emit(latestNews) // Emits the result of the request to the flow
            delay(refreshIntervalMs) // Suspends the coroutine for some time
        }
    }
}

// Interface that provides a way to make network requests with suspend functions
interface NewsApi {
    suspend fun fetchLatestNews(): List<ArticleHeadline>
}Flow.kt

ระบบจะเรียกใช้เครื่องมือสร้าง flow ภายในโครูทีน จึงได้รับประโยชน์จาก API แบบไม่พร้อมกันเดียวกัน แต่มีข้อจำกัดบางประการดังนี้

โฟลว์เป็นแบบตามลำดับ เนื่องจากโปรดิวเซอร์อยู่ในโครูทีน เมื่อเรียกใช้ฟังก์ชันระงับ โปรดิวเซอร์จะระงับจนกว่าฟังก์ชันระงับจะส่งคืน ในตัวอย่างนี้ โปรดิวเซอร์จะหยุดชั่วคราวจนกว่าfetchLatestNews คำขอเครือข่ายจะเสร็จสมบูรณ์ เมื่อนั้นระบบจึงจะส่งผลลัพธ์ไปยังสตรีม
flow บิลเดอร์ ผู้ผลิตไม่สามารถ emit ค่าจาก CoroutineContext อื่น ดังนั้น อย่าเรียกใช้ emit ในCoroutineContextอื่นโดยการสร้างโครูทีนใหม่หรือใช้บล็อกโค้ด withContext คุณสามารถใช้เครื่องมือสร้างโฟลว์อื่นๆ เช่น callbackFlow ในกรณีเหล่านี้ได้

การแก้ไขสตรีม

ตัวกลางสามารถใช้ตัวดำเนินการกลางเพื่อแก้ไขสตรีมของ ข้อมูลโดยไม่ต้องใช้ค่า ตัวดำเนินการเหล่านี้เป็นฟังก์ชันที่เมื่อ ใช้กับสตรีมข้อมูล จะตั้งค่าเชนของการดำเนินการที่ จะไม่ดำเนินการจนกว่าจะมีการใช้ค่าในอนาคต ดูข้อมูลเพิ่มเติมเกี่ยวกับ โอเปอเรเตอร์ระดับกลางได้ใน เอกสารอ้างอิง Flow

ในตัวอย่างด้านล่าง เลเยอร์ที่เก็บจะใช้อันดับกลาง map เพื่อเปลี่ยนรูปแบบข้อมูลที่จะแสดงใน View

class NewsRepository(
    private val newsRemoteDataSource: NewsRemoteDataSource,
    private val userData: UserData
) {
    /**
     * Returns the favorite latest news applying transformations on the flow.
     * These operations are lazy and don't trigger the flow. They just transform
     * the current value emitted by the flow at that point in time.
     */
    val favoriteLatestNews: Flow<List<ArticleHeadline>> =
        newsRemoteDataSource.latestNews
            // Intermediate operation to filter the list of favorite topics
            .map { news -> news.filter { userData.isFavoriteTopic(it) } }
            // Intermediate operation to save the latest news in the cache
            .onEach { news -> saveInCache(news) }
}Flow.kt

สามารถใช้ตัวดำเนินการระดับกลางต่อกันได้ ซึ่งจะสร้างห่วงโซ่ ของการดำเนินการที่จะดำเนินการแบบเลื่อนเวลาเมื่อมีการปล่อยรายการลงในโฟลว์ โปรดทราบว่าการใช้ตัวดำเนินการระดับกลางกับสตรีมเพียงอย่างเดียว ไม่ได้เริ่มการรวบรวมโฟลว์

การรวบรวมจากโฟลว์

ใช้โอเปอเรเตอร์เทอร์มินัลเพื่อทริกเกอร์โฟลว์ให้เริ่มฟังค่า หากต้องการรับค่าทั้งหมดในสตรีมเมื่อมีการปล่อยออกมา ให้ใช้ collect ดูข้อมูลเพิ่มเติมเกี่ยวกับโอเปอเรเตอร์เทอร์มินัลได้ในเอกสารประกอบเกี่ยวกับโฟลว์อย่างเป็นทางการ

เนื่องจาก collect เป็นฟังก์ชันระงับ จึงต้องเรียกใช้ภายใน โครูทีน โดยจะใช้ lambda เป็นพารามิเตอร์ที่เรียกใช้กับค่าใหม่ทุกค่า เนื่องจากเป็นฟังก์ชันระงับ โครูทีนที่เรียก collect อาจระงับจนกว่าจะปิดโฟลว์

จากตัวอย่างก่อนหน้า นี่คือการใช้งาน ViewModel อย่างง่ายที่ใช้ข้อมูลจากชั้นที่เก็บข้อมูล

class LatestNewsViewModel(
    private val newsRepository: NewsRepository
) : ViewModel() {

    init {
        viewModelScope.launch {
            // Trigger the flow and consume its elements using collect
            newsRepository.favoriteLatestNews.collect { favoriteNews ->
                // Update UI with the latest favorite news
            }
        }
    }
}Flow.kt

การรวบรวมโฟลว์จะทริกเกอร์โปรดิวเซอร์ที่รีเฟรชข่าวสารล่าสุด และปล่อยผลลัพธ์ของคำขอเครือข่ายในช่วงเวลาที่กำหนด เนื่องจาก Producer จะยังคงทำงานอยู่เสมอด้วยลูป while(true) สตรีม ของข้อมูลจะปิดเมื่อ ViewModel ถูกล้างและ viewModelScope ถูกยกเลิก

การรวบรวมโฟลว์อาจหยุดลงด้วยเหตุผลต่อไปนี้

ระบบจะยกเลิกโครูทีนที่รวบรวม ดังที่แสดงในตัวอย่างก่อนหน้า ซึ่งจะเป็นการหยุดผู้ผลิตที่เกี่ยวข้องด้วย
ผู้ผลิตส่งรายการเสร็จสิ้น ในกรณีนี้ สตรีมข้อมูล จะปิดและโครูทีนที่เรียก collect จะกลับมาดำเนินการต่อ

Flow จะเป็นแบบ cold และ lazy เว้นแต่จะระบุด้วยตัวดำเนินการระดับกลางอื่นๆ ซึ่งหมายความว่าโค้ดของผู้ผลิตจะได้รับการเรียกใช้ทุกครั้งที่มีการเรียกใช้โอเปอเรเตอร์เทอร์มินัลในโฟลว์ ในตัวอย่างก่อนหน้า การมีตัวรวบรวมโฟลว์หลายตัวจะทำให้แหล่งข้อมูลดึงข้อมูล ข่าวสารล่าสุดหลายครั้งในช่วงเวลาที่กำหนดต่างๆ หากต้องการเพิ่มประสิทธิภาพและแชร์โฟลว์เมื่อผู้บริโภคหลายรายรวบรวมข้อมูลพร้อมกัน ให้ใช้โอเปอเรเตอร์ shareIn

การตรวจหาข้อยกเว้นที่ไม่คาดคิด

การติดตั้งใช้งาน Producer อาจมาจากไลบรารีของบุคคลที่สาม ซึ่งหมายความว่าฟังก์ชันนี้อาจทำให้เกิดข้อยกเว้นที่ไม่คาดคิด หากต้องการจัดการข้อยกเว้นเหล่านี้ ให้ใช้ตัวดำเนินการกลาง catch

class LatestNewsViewModel(
    private val newsRepository: NewsRepository
) : ViewModel() {

    init {
        viewModelScope.launch {
            newsRepository.favoriteLatestNews
                // Intermediate catch operator. If an exception is thrown,
                // catch and update the UI
                .catch { exception -> notifyError(exception) }
                .collect { favoriteNews ->
                    // Update UI with the latest favorite news
                }
        }
    }
}Flow.kt

ในตัวอย่างก่อนหน้า เมื่อเกิดข้อยกเว้น ระบบจะไม่เรียกใช้ collect Lambda เนื่องจากยังไม่ได้รับสินค้าใหม่

นอกจากนี้ยังcatchรายการemitลงในโฟลว์ได้ด้วย ที่เก็บตัวอย่าง เลเยอร์อาจemitค่าที่แคชไว้แทน

class NewsRepository(
    // ...
) {
    val favoriteLatestNews: Flow<List<ArticleHeadline>> =
        newsRemoteDataSource.latestNews
            .map { news -> news.filter { userData.isFavoriteTopic(it) } }
            .onEach { news -> saveInCache(news) }
            // If an error happens, emit the last cached values
            .catch { exception -> emit(lastCachedNews()) }
}Flow.kt

ในตัวอย่างนี้ เมื่อเกิดข้อยกเว้น ระบบจะเรียกใช้ collect Lambda เนื่องจากมีการส่งรายการใหม่ไปยังสตรีมเนื่องจาก ข้อยกเว้น

การดำเนินการใน CoroutineContext อื่น

โดยค่าเริ่มต้น ตัวสร้าง flow จะทำงานใน CoroutineContext ของโครูทีนที่รวบรวมจากตัวสร้างนั้น และตามที่กล่าวไว้ก่อนหน้านี้ ตัวสร้างจะไม่สามารถ emit ค่าจาก CoroutineContext อื่นได้ ลักษณะการทำงานนี้อาจไม่เป็นที่ต้องการในบางกรณี เช่น ในตัวอย่างที่ใช้ตลอดทั้งหัวข้อนี้ เลเยอร์ที่เก็บไม่ควรดำเนินการกับ Dispatchers.Main ที่ viewModelScope ใช้

หากต้องการเปลี่ยน CoroutineContext ของโฟลว์ ให้ใช้ตัวดำเนินการกลาง flowOn flowOn จะเปลี่ยนCoroutineContextของโฟลว์ต้นทาง ซึ่งหมายความว่า ผู้ผลิตและโอเปอเรเตอร์ระดับกลางที่ใช้ก่อน (หรือเหนือ) flowOn โฟลว์ดาวน์สตรีม (โอเปอเรเตอร์ระดับกลางหลังจาก flowOn พร้อมกับผู้บริโภค) จะไม่ได้รับผลกระทบและจะดำเนินการในCoroutineContextที่ใช้collectจากโฟลว์ หากมีflowOnผู้ให้บริการหลายราย ผู้ให้บริการแต่ละรายจะเปลี่ยนต้นทางจากตำแหน่งปัจจุบัน

class NewsRepository(
    private val newsRemoteDataSource: NewsRemoteDataSource,
    private val userData: UserData,
    private val defaultDispatcher: CoroutineDispatcher
) {
    val favoriteLatestNews: Flow<List<ArticleHeadline>> =
        newsRemoteDataSource.latestNews
            .map { news -> // Executes on the default dispatcher
                news.filter { userData.isFavoriteTopic(it) }
            }
            .onEach { news -> // Executes on the default dispatcher
                saveInCache(news)
            }
            // flowOn affects the upstream flow ↑
            .flowOn(defaultDispatcher)
            // the downstream flow ↓ is not affected
            .catch { exception -> // Executes in the consumer's context
                emit(lastCachedNews())
            }
}Flow.kt

เมื่อใช้โค้ดนี้ โอเปอเรเตอร์ onEach และ map จะใช้ defaultDispatcher ส่วนโอเปอเรเตอร์ catch และผู้บริโภคจะดำเนินการใน Dispatchers.Main ที่ viewModelScope ใช้

เนื่องจากเลเยอร์แหล่งข้อมูลทำงาน I/O คุณจึงควรใช้ Dispatcher ที่เพิ่มประสิทธิภาพสำหรับการดำเนินการ I/O

class NewsRemoteDataSource(
    // ...
    private val ioDispatcher: CoroutineDispatcher
) {

    val latestNews: Flow<List<ArticleHeadline>> = flow {
        // Executes on the IO dispatcher
        // ...
    }
        .flowOn(ioDispatcher)
}Flow.kt

Flow ในไลบรารี Jetpack

Flow ได้รับการผสานรวมเข้ากับไลบรารี Jetpack หลายรายการ และเป็นที่นิยมในหมู่ ไลบรารีของบุคคลที่สามสำหรับ Android Flow เหมาะสำหรับการอัปเดตข้อมูลแบบเรียลไทม์ และสตรีมข้อมูลที่ไม่มีที่สิ้นสุด

คุณใช้ Flow with Room เพื่อรับการแจ้งเตือนเกี่ยวกับการเปลี่ยนแปลงในฐานข้อมูลได้ เมื่อใช้ออบเจ็กต์การเข้าถึงข้อมูล (DAO) ให้ส่งคืนประเภท Flow เพื่อรับข้อมูลอัปเดตแบบเรียลไทม์

@Dao
abstract class ExampleDao {
    @Query("SELECT * FROM Example")
    abstract fun getExamples(): Flow<List<Example>>
}Flow.kt

ทุกครั้งที่มีการเปลี่ยนแปลงในExampleตาราง ระบบจะส่งรายการใหม่ พร้อมกับรายการใหม่ในฐานข้อมูล

แปลง API ที่อิงตามการเรียกกลับเป็นโฟลว์

callbackFlow เป็นเครื่องมือสร้างโฟลว์ที่ช่วยให้คุณแปลง API ที่อิงตามการเรียกกลับเป็นโฟลว์ได้ ตัวอย่างเช่น Firebase Firestore Android API ใช้การเรียกกลับ

หากต้องการแปลง API เหล่านี้เป็นโฟลว์และรอรับการอัปเดตฐานข้อมูล Firestore คุณ สามารถใช้โค้ดต่อไปนี้ได้

class FirestoreUserEventsDataSource(
    private val firestore: FirebaseFirestore
) {
    // Method to get user events from the Firestore database
    fun getUserEvents(): Flow<UserEvents> = callbackFlow {

        // Reference to use in Firestore
        var eventsCollection: CollectionReference? = null
        try {
            eventsCollection = FirebaseFirestore.getInstance()
                .collection("collection")
                .document("app")
        } catch (e: Throwable) {
            // If Firebase cannot be initialized, close the stream of data
            // flow consumers will stop collecting and the coroutine will resume
            close(e)
        }

        // Registers callback to firestore, which will be called on new events
        val subscription = eventsCollection?.addSnapshotListener { snapshot, _ ->
            if (snapshot == null) { return@addSnapshotListener }
            // Sends events to the flow! Consumers will get the new events
            try {
                trySend(snapshot.getEvents())
            } catch (e: Throwable) {
                // Event couldn't be sent to the flow
            }
        }

        // The callback inside awaitClose will be executed when the flow is
        // either closed or cancelled.
        // In this case, remove the callback from Firestore
        awaitClose { subscription?.remove() }
    }
}

callbackFlow ต่างจากflow Builder ตรงที่อนุญาตให้ส่งค่าจาก CoroutineContext อื่นด้วยฟังก์ชัน send หรือภายนอกโครูทีนด้วยฟังก์ชัน trySend

ภายใน callbackFlow จะใช้แชแนล ซึ่งมีแนวคิดคล้ายกับคิวที่บล็อก มาก ช่องจะได้รับการกำหนดค่าด้วยความจุ ซึ่งเป็นจำนวนองค์ประกอบสูงสุด ที่บัฟเฟอร์ได้ ช่องที่สร้างใน callbackFlow มีความจุเริ่มต้น ที่ 64 องค์ประกอบ เมื่อคุณพยายามเพิ่มองค์ประกอบใหม่ลงในช่องแบบเต็ม send จะระงับผู้ผลิตจนกว่าจะมีพื้นที่สำหรับองค์ประกอบใหม่ ในขณะที่ trySend จะไม่เพิ่มองค์ประกอบลงในช่องและจะแสดงข้อผิดพลาด false ทันที

trySend จะเพิ่มองค์ประกอบที่ระบุลงในช่องทันที ก็ต่อเมื่อการดำเนินการนี้ไม่ละเมิดข้อจำกัดด้านความจุ จากนั้นจะแสดงผลลัพธ์ที่ สำเร็จ