Kotlin Flow Cheat Sheet: Phần 1 - Channel

Khi bạn đã làm quen với Kotlin Flows, có thể bạn đã nắm vững các khái niệm cơ bản. Tuy nhiên, nếu bạn chưa từng sử dụng Channel, có thể bạn chưa phân biệt được giữa merge, combine và zip, hay chưa hiểu rõ về SharedFlow và StateFlow, cũng như cách sử dụng chúng.

Bài viết này tổng hợp những kiến thức thiết yếu mà tôi đã tích lũy được trong quá trình làm việc với Kotlin Flow. Tài liệu này được thiết kế như một bảng tham chiếu hữu ích nhằm giúp bạn giải quyết các tình huống phức tạp trong lập trình coroutine.

Trong phần này, chúng ta sẽ đi sâu vào Channel, tìm hiểu cách hoạt động của nó và áp dụng nó vào các trường hợp thực tế để giao tiếp một cách an toàn và hiệu quả giữa các coroutine.

Tổng Quan Về Stream

Hot Streams

• Ví dụ: Channel, Collections (như List, Set, etc.).
• Phát ngay lập tức: Bắt đầu phát giá trị bất kể có subscriber hay không.
• Lưu trữ phần tử: Không cần tính toán lại, tất cả subscriber đều nhận cùng một chuỗi giá trị.

Cold Streams

• Ví dụ: Sequence, Flow.
• Bắt đầu theo yêu cầu: Chỉ bắt đầu phát giá trị khi subscriber chủ động đăng ký.
• Phát độc lập: Mỗi subscriber nhận chuỗi giá trị riêng của mình, không có phần tử nào được lưu trữ.

Khám Phá Channel

Nguyên Tắc Chính

• Là một hot stream.
• Đảm bảo không có xung đột (không có vấn đề với trạng thái chia sẻ) và công bằng, hữu ích khi nhiều coroutine cần liên lạc với nhau.
• Hỗ trợ bất kỳ số lượng sender và receiver.
• Mỗi giá trị gửi tới channel chỉ được nhận một lần.
• Nếu có nhiều receiver subscribe, phần tử sẽ được phân bổ công bằng theo hàng đợi FIFO.

Chức Năng Chính

• Channel có 2 hàm suspend: send và receive.
• receive sẽ bị suspend nếu không có phần tử nào trong channel và chờ đến khi có phần tử mới.
• send sẽ bị suspend nếu channel đã đạt tới capacity.
• Cũng có phiên bản không bị suspend là trySend và tryReceive, trả về một ChannelResult để cho biết thao tác thành công hay không.
• Cần đóng channel thủ công sau khi hoàn tất gửi dữ liệu: myChannel.close(). Nếu không, receive sẽ đợi mãi mãi.

Các Loại Channel Capacity

val myChannel = Channel<Int>(capacity = 3)

• Channel.UNLIMITED: Buffer không giới hạn, send không bao giờ bị suspend.
• Channel.BUFFERED: Buffer capacity là 64 (mặc định), có thể ghi đè bằng thuộc tính hệ thống.
• Channel.RENDEZVOUS: Buffer capacity là 0 (hành vi mặc định), receiver chỉ nhận được dữ liệu nếu đã subscribe trước khi dữ liệu phát.
• Channel.CONFLATED: Buffer capacity là 1, phần tử mới thay thế phần tử cũ nhất.
• Giá trị int bất kỳ: Buffer sẽ có capacity tương ứng.

Xử Lý Lỗi Tràn Buffer

• Tham số onBufferOverflow kiểm soát hành vi khi buffer đầy với 3 lựa chọn:
  • BufferOverflow.SUSPEND: (hành vi mặc định) tạm dừng send khi buffer đầy.
  • BufferOverflow.DROP_OLDEST: loại bỏ phần tử cũ nhất khi buffer đầy.
  • BufferOverflow.DROP_LATEST: loại bỏ phần tử mới nhất khi buffer đầy.

Tạo Channel Tự Đọng Đóng

• Coroutine builder produce sẽ tự động đóng channel khi builder coroutine kết thúc.

suspend fun myFunction() = coroutineScope {
    val channel = produce {
        // phát các giá trị ở đây và không cần gọi close() khi kết thúc
    }
}

Tự Động Dọn Dẹp Khi Một Phần Tử Không Thể Xử Lý

val myChannel = Channel(
    capacity,
    onUnderliveredElement = { /* các tác vụ dọn dẹp ở đây */ }
)

Trường Hợp Sử Dụng: Kích Hoạt Làm Mới

Trong Android, một trường hợp phổ biến cho channel là kích hoạt làm mới khi màn hình được refresh (ví dụ: pull to refresh hoặc button retry).

Dolược mã bên dưới cho thấy cách lấy dữ liệu từ API khi chúng ta subscribe flow lần đầu tiên hoặc khi kích hoạt làm mới.

interface ApiService {
    suspend fun fetchData(): List<String>
}

class FetchDataUseCase @Inject constructor (
    private val apiService: ApiService
) {
    private val refreshChannel = Channel<Unit>(
        capacity = 1,
        onBufferOverflow = BufferOverflow.DROP_LATEST
    )

    val dataState: Flow<FetchDataState> =
        refreshChannel
            .consumeAsFlow()
            .onStart { emit(Unit) }
            .map { fetchData() }

    fun refresh() {
        refreshChannel.trySend(Unit)
    }

    private suspend fun fetchData(): FetchDataState =
        try {
            val data = apiService.fetchData()
            FetchDataState.Success(data)
        } catch (e: Exception) {
            FetchDataState.Error(e.message ?: "An error occurred")
        }

    sealed interface FetchDataState {
        data object Loading : FetchDataState
        data class Success(val data: List<String>) : FetchDataState
        data class Error(val message: String) : FetchDataState
    }
}

Cảm ơn bạn đã theo dõi. Hãy chờ đón các phần tiếp theo nhé!

Tài Liệu Tham Khảo

• Xem thêm trên Medium

🔔 Blog: henrytechie.com

☕️ Facebook: Henry Techie

☁️ TikTok: @henrytechie
source: viblo