Flow — คืออะไร, สตรีมแบบ cold และ hot ในโครูทีน Kotlin

Flow ใน Kotlin คืออะไร?

Flow เป็นประเภทจากแพ็กเกจ kotlinx.coroutines.flow ที่แสดงถึงสตรีมข้อมูลแบบอะซิงโครนัส cold โดยพื้นฐานแล้ว Flow คือลำดับโครูทีนที่ปล่อยค่าผ่านฟังก์ชัน emit() และสิ้นสุดด้วยความสำเร็จหรือข้อยกเว้น การรวบรวมสตรีมดำเนินการผ่านโอเปอเรเตอร์ปลายทาง collect() ซึ่งเป็นฟังก์ชัน suspend

ความหมายแบบ Cold

สตรีม cold หมายความว่าโค้ดภายในตัวสร้าง flow จะทำงานใหม่สำหรับผู้สมัครสมาชิกแต่ละคน Observable.fromIterable ใน RxJava มีพฤติกรรมคล้ายกัน: ผู้สมัครสมาชิกใหม่ได้รับค่าทั้งหมดตั้งแต่ต้น ใน Flow สิ่งนี้ถูกนำไปใช้ผ่านฟังก์ชัน suspend collect ซึ่งบล็อกโครูทีนตลอดระยะเวลาการรวบรวมข้อมูล

ตัวสร้าง Flow

Kotlin มีหลายวิธีในการสร้าง Flow: flow { } — โครงสร้างพื้นฐานด้วย emit(), flowOf(vararg values) — สำหรับชุดค่าคงที่, .asFlow() — ส่วนขยายสำหรับคอลเลกชันและ Sequence ตัวสร้างทั้งหมดเป็นแบบ cold — ข้อมูลจะถูกสร้างเมื่อเรียกโอเปอเรเตอร์ปลายทางเท่านั้น

สตรีม Cold และ Hot

การแบ่งระหว่างสตรีม cold และ hot เป็นแนวคิดหลักของการเขียนโปรแกรมเชิงรับ สตรีม cold (Flow, Observable) เริ่มสร้างข้อมูลเมื่อมีการสมัครสมาชิก สตรีม hot (Channel, SharedFlow) ปล่อยข้อมูลอย่างอิสระ — ผู้สมัครสมาชิกได้รับเฉพาะสิ่งที่เกิดขึ้นหลังจากการสมัคร โดยไม่มีจุดเริ่มต้นของลำดับ

SharedFlow เป็น Flow แบบ hot ที่สามารถมีผู้สมัครสมาชิกหลายคนและเล่นค่าล่าสุดซ้ำเมื่อมีการกำหนดค่า replay SharedFlow เหมาะสำหรับเหตุการณ์ (การแจ้งเตือนแบบครั้งเดียว) StateFlow เป็นรูปแบบที่มีค่าสถานะคงที่ แคชค่าล่าสุดสำหรับผู้สมัครสมาชิกใหม่

ChannelFlow ใช้ Channel ภายใน รวมคุณสมบัติของ Flow และ Channel รองรับการบัฟเฟอร์และ backpressure ผ่าน capacity ChannelFlow มีประโยชน์เมื่อแปลง API callback เป็นสตรีมเชิงรับ ซึ่งค่าถูกปล่อยจากโครูทีนต่างๆ

การแปลงระหว่าง cold และ hot

ในการแปลง cold Flow เป็น hot SharedFlow ใช้โอเปอเรเตอร์ shareIn(scope, started, replay) พารามิเตอร์ started ควบคุมช่วงเวลาเริ่มต้น: SharingStarted.WhileSubscribed() — ทำงานเมื่อมีผู้สมัครสมาชิก, Lazily — เริ่มที่ผู้สมัครสมาชิกแรก, Eagerly — เริ่มทันที การแปลงย้อนกลับ — hot เป็น cold: StateFlow.asFlow() คืนค่า Flow แบบ cold ที่เมื่อ collect จะปล่อยค่าปัจจุบันของ StateFlow สะดวกสำหรับการทดสอบ

โอเปอเรเตอร์ Flow

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

โอเปอเรเตอร์ map แปลงค่าแต่ละค่าของสตรีมผ่านการแปลงแบบอะซิงโครนัสหรือซิงโครนัส filter ผ่านเฉพาะค่าที่ตรงตามเงื่อนไข catch จับข้อยกเว้นก่อนโอเปอเรเตอร์ปลายทางและอนุญาตให้กู้คืนสตรีม flatMapLatest ยกเลิกการปล่อยครั้งก่อนเมื่อมีค่าใหม่เข้ามา — คล้ายกับ switchMap ใน Rx

โอเปอเรเตอร์ debounce ใน Flow ชะลอการเผยแพร่ค่าตามระยะเวลาที่กำหนด หากมีค่าใหม่เข้ามาในช่วงเวลานี้ ตัวจับเวลาจะถูกรีเซ็ต ใน Android มีการใช้ debounce สำหรับการค้นหา: คำขอจะถูกส่งหลังจากหยุดชั่วคราว 300-400 ms ซึ่งลดการเรียก API ลง 3-5 เท่า

โอเปอเรเตอร์ปลายทาง

นอกเหนือจาก collect() แล้ว Flow ยังรองรับโอเปอเรเตอร์ปลายทางอื่นๆ: toList() รวบรวมค่าทั้งหมดในรายการ — มีประโยชน์สำหรับการทดสอบ, first() คืนค่าองค์ประกอบแรกและยกเลิกสตรีม, single() คาดว่ามีหนึ่งองค์ประกอบพอดี fold(initial) สะสมค่าผ่านฟังก์ชันที่ส่งผ่าน โอเปอเรเตอร์ปลายทางทั้งหมดเป็นฟังก์ชัน suspend และต้องเรียกภายในโครูทีนหรือฟังก์ชัน suspend อื่น

ตัวอย่างโค้ด Flow

ตัวอย่างแรก — Flow พื้นฐานที่สร้างตัวเลขด้วยการแปลงผ่านโอเปอเรเตอร์ map:

val numberFlow = flow {
    for (i in 1..5) {
        delay(500)
        emit(i)
    }
}

scope.launch {
    numberFlow
        .map { "จำนวน: $it" }
        .collect { value ->
            println(value)
        }
}

ตัวอย่างที่สอง — การแปลงสตรีมด้วยการกรองและการจัดการข้อผิดพลาดผ่าน catch:

flow {
    emit("data1")
    emit("data2")
    throw RuntimeException("network error")
}
    .catch { e ->
        emit("fallback_data")
    }
    .collect { value ->
        println(value)
    }

ตัวอย่างที่สาม — การใช้ StateFlow ใน ViewModel สำหรับ UI เชิงรับใน Jetpack Compose:

class SearchViewModel : ViewModel() {
    private val _query = MutableStateFlow("")
    val results: StateFlow<List<Result>> = _query
        .debounce(300)
        .flatMapLatest { query ->
            repository.search(query)
        }
        .catch { emit(emptyList()) }
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), emptyList())

    fun onQueryChanged(query: String) {
        _query.value = query
    }
}

StateFlow และ SharedFlow

StateFlow เป็น Flow แบบ hot ที่มีค่าปัจจุบันเดียว แคชค่าล่าสุดและส่งต่อไปยังผู้สมัครสมาชิกใหม่ทันที StateFlow เป็นคอนเทนเนอร์ที่สังเกตได้สำหรับสถานะ รองรับการเปรียบเทียบ equals — หากค่าใหม่ตรงกับค่าปัจจุบัน จะไม่มีการปล่อยเกิดขึ้น Jetpack Compose ใช้ StateFlow ผ่าน collectAsState()

SharedFlow เป็น Flow แบบ hot ที่ยืดหยุ่นกว่าโดยไม่มีค่าเริ่มต้นบังคับ SharedFlow ถูกกำหนดค่าผ่าน replay (จำนวนค่าสำหรับผู้สมัครสมาชิกใหม่), extraBufferCapacity (บัฟเฟอร์เกิน replay), และ onBufferOverflow (กลยุทธ์เมื่อล้น) SharedFlow เหมาะอย่างยิ่งสำหรับเหตุการณ์แบบครั้งเดียว: การนำทาง, Snackbar, การวิเคราะห์

Flow ในสถาปัตยกรรม Android ได้รับคำแนะนำจาก Google เป็นแหล่งข้อมูลหลัก (ชั้น: Repository → UseCase → ViewModel) LiveData มีความยืดหยุ่นน้อยกว่า Flow: Flow รองรับโครูทีน โอเปอเรเตอร์ backpressure และทำงานนอกชั้น UI การโยกย้ายจาก LiveData ไปยัง Flow เป็นแนวทางปฏิบัติมาตรฐานในโปรเจกต์ Android สมัยใหม่

เมื่อใช้ Flow ใน ViewModel การเลือกประเภทที่ถูกต้องเป็นสิ่งสำคัญ StateFlow เหมาะสำหรับสถานะ UI ที่ต้องอยู่รอดเมื่อหมุนหน้าจอ SharedFlow เหมาะสำหรับเหตุการณ์ที่ไม่สามารถประมวลผลซ้ำได้ — เช่น การนำทาง Flow ที่มี collect() ใน lifecycleScope ให้การควบคุมสูงสุดเหนือบริบทการดำเนินการ แต่ต้องยกเลิกด้วยตนเองเมื่อออกจากหน้าจอ

การทดสอบ Flow ทำผ่าน kotlinx-coroutines-test ไลบรารีมี TestDispatcher — เวลาเสมือนที่ช่วยเร่งความล่าช้า (delay) และควบคุมลำดับการดำเนินการของโครูทีน TestScope.runTest { } สร้างสภาพแวดล้อมแบบแยกส่วนสำหรับทดสอบ Flow โอเปอเรเตอร์ toList() มักใช้ในการทดสอบเพื่อรวบรวมค่า flow ทั้งหมดด้วยการหมดเวลา เพื่อตรวจสอบว่าสตรีมปล่อยลำดับข้อมูลที่ถูกต้อง

Flow ผสานรวมกับ Room (ไลบรารี Android สำหรับ DB) ได้ดี: เมธอด DAO สามารถคืนค่า Flow<List<Entity>> Room ปล่อยค่าใหม่โดยอัตโนมัติเมื่อมีการเปลี่ยนแปลงตาราง — UI อัปเดตโดยไม่ต้องมีทริกเกอร์ด้วยตนเอง สิ่งนี้ถูกนำไปใช้ผ่าน InvalidationTracker ซึ่งภายใต้ฝาใช้ Flow กับ callbackFlow วิธีการนี้ขจัดความจำเป็นของ LiveData และทำให้ชั้นข้อมูลมุ่งไปที่โครูทีนอย่างสมบูรณ์ Jetpack Compose ผ่าน collectAsState() สมัครสมาชิก StateFlow และวาดเฉพาะคอมโพเนนต์ที่ข้อมูลเปลี่ยนแปลง — ซึ่งให้ประสิทธิภาพที่ไม่สามารถบรรลุได้ด้วยสถาปัตยกรรมที่มุ่งเน้น LiveData DataStore (ตัวแทน SharedPreferences) ก็คืนค่า Flow<Preferences> เช่นกัน ให้การอ่านการตั้งค่าแอปพลิเคชันแบบ reacting โดยไม่ต้องใช้ทริกเกอร์อัปเดตด้วยตนเอง

Flow รองรับการสื่อสารระหว่างกระบวนการผ่าน kotlinx-coroutines-core บน JVM โดยไม่ต้องใช้ไลบรารีเพิ่มเติม ตัวอย่างเช่น ในแอปพลิเคชันเซิร์ฟเวอร์บน Ktor Flow สามารถแสดงถึงสตรีมข้อความ WebSocket ขาเข้า แต่ละข้อความถูกปล่อยเข้าสู่สตรีม ผ่านการกรองและการรวมผ่านโอเปอเรเตอร์ และผลลัพธ์ถูกส่งไปยังไคลเอนต์ วิธีการนี้แทนที่ไลบรารีเชิงรับอย่าง Reactor หรือ RxJava ในโปรเจกต์ Kotlin

ความเข้ากันได้ของ Flow กับโค้ด RxJava ที่มีอยู่มีให้โดยโมดูล kotlinx-coroutines-rx3 ฟังก์ชันส่วนขยาย Flow.asObservable() แปลง Flow เป็น Observable ของ RxJava 3 การแปลงย้อนกลับ — CompletableSource.asFlow(), Observable.asFlow() ซึ่งทำให้การโยกย้ายจาก RxJava ไปยังโครูทีนง่ายขึ้น: สามารถเขียนโปรเจกต์ใหม่เป็นขั้นตอน โดยปล่อยบางชั้นไว้บน RxJava เมื่อแปลง ต้องพิจารณาความแตกต่างในความหมาย cold/hot: Observable สามารถเป็นได้ทั้ง cold และ hot Flow จะเป็น cold เสมอสำหรับ Flow ทั่วไปและ hot สำหรับ SharedFlow

การจัดการข้อผิดพลาดและการทดสอบ Flow

การจัดการข้อผิดพลาดใน Flow มีลักษณะเฉพาะ: หากข้อยกเว้นเกิดขึ้นภายในตัวสร้าง flow ก่อนโอเปอเรเตอร์ปลายทาง มันจะถูกส่งต่อไปยัง catch หากข้อยกเว้นเกิดขึ้นในโอเปอเรเตอร์หลังจากตัวสร้าง catch หลังจากโอเปอเรเตอร์นั้นจะจับมัน retryWhen อนุญาตให้ลองสมัครสมาชิกอีกครั้งโดยมีเงื่อนไข: ลองใหม่เมื่อข้อผิดพลาดเครือข่ายสูงสุด 3 ครั้ง แต่ไม่ลองใหม่เมื่อ CancellationException Flow ขจัดข้อผิดพลาดที่ขึ้นกับสถานะเนื่องจากไม่เก็บสถานะ — ซึ่งทำให้การดีบักง่ายขึ้นเมื่อเทียบกับ Observable ที่ Subject เก็บสถานะภายใน

การทดสอบ Flow ด้วย kotlinx-coroutines-test ใช้ TestDispatcher เพื่อจำลองความล่าช้า Turbine เป็นไลบรารีชุมชนยอดนิยมสำหรับทดสอบ Flow: test { } เริ่ม Flow, awaitItem() รอค่าถัดไป, awaitComplete() รอความสมบูรณ์ Turbine เพิ่มการหมดเวลาเริ่มต้น ป้องกันการค้างของการทดสอบ สำหรับการทดสอบ StateFlow ให้ใช้ .testIn(scope) ด้วยการตรวจสอบค่าตามลำดับเวลา