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

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

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

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

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

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

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

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

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

หากต้องการสร้างโฟลว์ ให้ใช้ 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 ตัวสร้างจะดำเนินการภายในโครูทีน จึงได้รับประโยชน์จาก API แบบไม่พร้อมกันเดียวกัน แต่มีข้อจำกัดบางประการดังนี้

  • โฟลว์เป็นแบบตามลำดับ เนื่องจาก Producer อยู่ในโครูทีน เมื่อเรียกใช้ฟังก์ชันระงับ Producer จะระงับจนกว่าฟังก์ชันระงับจะส่งคืน ในตัวอย่างนี้ โปรดิวเซอร์จะระงับจนกว่า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) }
}

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

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

ใช้โอเปอเรเตอร์เทอร์มินัลเพื่อทริกเกอร์โฟลว์ให้เริ่มฟังค่า หากต้องการรับค่าทั้งหมดในสตรีมเมื่อมีการปล่อยออกมา ให้ใช้ 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
            }
        }
    }
}

การรวบรวมโฟลว์จะทริกเกอร์โปรดิวเซอร์ที่รีเฟรชข่าวสารล่าสุด และปล่อยผลลัพธ์ของคำขอเครือข่ายในช่วงเวลาที่กำหนด เนื่องจาก 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
                }
        }
    }
}

ในตัวอย่างก่อนหน้า เมื่อเกิดข้อยกเว้น ระบบจะไม่เรียกใช้ 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()) }
}

ในตัวอย่างนี้ เมื่อเกิดข้อยกเว้น ระบบจะเรียกใช้ 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())
            }
}

เมื่อใช้โค้ดนี้ โอเปอเรเตอร์ 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)
}

โฟลว์ในไลบรารี Jetpack และแหล่งข้อมูล

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

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

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

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

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

class RemoteMessagesDataSource(
    private val firestore: FirebaseFirestore
) {
    // Returns a Flow emitting updated message lists when collection changes
    val messages: Flow<List<Message>> =
        firestore.collection("messages").dataObjects()
}

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

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

callbackFlow เป็นเครื่องมือสร้างโฟลว์ที่ช่วยให้คุณแปลง API ที่อิงตาม Callback เป็นโฟลว์ได้ ซึ่งมีประโยชน์อย่างยิ่งเมื่อใช้ Listener เหตุการณ์หลายรายการใน SDK เช่น Cloud Firestore หรือ การกำหนดค่าระยะไกลของ Firebase

หากต้องการแปลง API ของ Listener ที่อิงตามการเรียกกลับเป็นโฟลว์ เช่น การฟังการอัปเดตฐานข้อมูล 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 จะเพิ่มองค์ประกอบที่ระบุลงในช่องทันที ก็ต่อเมื่อการดำเนินการนี้ไม่ละเมิดข้อจำกัดด้านความจุ จากนั้นจะแสดงผลลัพธ์ที่ สำเร็จ

แหล่งข้อมูลเพิ่มเติมเกี่ยวกับโฟลว์