Channel: bản chất, các loại kênh và coroutine trong Kotlin

Channel trong Kotlin là gì?

Channel về mặt khái niệm tương tự như BlockingQueue từ Java, nhưng với các hàm suspend send() và receive() thay vì put() và take() chặn. Một nhà phát triển Kotlin sử dụng Channel để tổ chức trao đổi dữ liệu giữa các coroutine mà không cần đồng bộ qua bộ nhớ dùng chung. Kênh đảm bảo phân phối có thứ tự — thứ tự gửi khớp với thứ tự nhận.

Tạo kênh

Để tạo Channel, hàm factory Channel<T>(capacity) được gọi. Tham số capacity xác định loại kênh: RENDEZVOUS (0), UNLIMITED (Int.MAX_VALUE), CONFLATED (-1) hoặc một số cụ thể. Loại phần tử T được chỉ định qua generics. Đóng kênh qua close() báo hiệu rằng sẽ không có phần tử mới nào nữa.

Send và Receive

send(value) là một hàm suspend tạm dừng coroutine gửi nếu kênh đầy. receive() là một hàm suspend tạm dừng người nhận nếu kênh trống. Các lựa chọn thay thế trySend() và tryReceive() là các phiên bản không chặn trả về Boolean hoặc null khi thao tác không khả thi. Chúng hữu ích trong các ngữ cảnh không suspend.

Các loại Channel trong kotlinx.coroutines

Kotlin cung cấp bốn biến thể Channel qua dung lượng bộ đệm: Rendezvous (dung lượng 0), Buffered (dung lượng N), Conflated (dung lượng 1, ghi đè) và Unlimited (dung lượng Int.MAX_VALUE). Mỗi loại giải quyết nhiệm vụ riêng, từ đồng bộ nghiêm ngặt đến đệm dữ liệu hàng loạt.

Rendezvous Channel là nghiêm ngặt nhất: send() bị chặn cho đến khi receive() được gọi trong một coroutine khác. Về bản chất, đây là điểm hẹn của hai coroutine. Lý tưởng cho bắt tay nghiêm ngặt khi người gửi phải đợi người nhận xử lý phần tử. Mất dữ liệu được loại trừ — send không hoàn thành cho đến khi receive được thực thi.

Conflated Channel chỉ lưu trữ giá trị cuối cùng được gửi. Nếu người gửi đặt phần tử mới trước khi người nhận lấy phần tử cũ, phần tử cũ bị loại bỏ. Conflated Channel hữu ích cho trạng thái UI: nếu người dùng thay đổi thanh trượt nhanh chóng, các giá trị trung gian có thể bị loại bỏ và chỉ xử lý giá trị cuối cùng.

Producer-Consumer với kênh

Mẫu Producer-Consumer cổ điển trên Channel được triển khai qua các coroutine song song. Producer gọi send(value) trong một vòng lặp, consumer gọi receive(value). Producer và consumer có thể hoạt động trên các Dispatchers khác nhau: producer trên Dispatchers.IO, consumer trên Dispatchers.Main. Channel tự động đồng bộ truy cập mà không cần Lock hay synchronized.

Fan-out — nhiều consumer trên một kênh duy nhất. Mỗi phần tử đi đến đúng một consumer (phân phối round-robin). Fan-in — nhiều producer ghi vào một kênh duy nhất. Các coroutine gửi cạnh tranh để gửi, nhưng thứ tự phần tử được bảo toàn. Cả hai kịch bản đều không yêu cầu đồng bộ bổ sung.

Produce là một coroutine builder tạo kênh với tự động đóng. Hàm produce { } trả về ReceiveChannel — một kênh chỉ đọc cho consumer. Bên trong builder, send() gửi dữ liệu, và khi khối hoàn thành hoặc có ngoại lệ, kênh tự động đóng, ngăn rò rỉ.

Select và ghép kênh

Thư viện kotlinx.coroutines cung cấp select — một biểu thức chờ kênh hoàn thành đầu tiên trong số nhiều lựa chọn thay thế. Select cho phép ghép kênh nhiều kênh: ví dụ, chờ dữ liệu từ hai nguồn và xử lý nguồn phản hồi đầu tiên. Cú pháp — select<T> { channel1.onReceive { } channel2.onReceive { } }. Đây là một giải pháp thay thế cho toán tử amb trong Rx.

Ví dụ mã Channel

Ví dụ đầu tiên là Rendezvous Channel đơn giản nơi người gửi chờ nhận:

val channel = Channel<String>()

scope.launch {
    channel.send("Hello")
    println("Đã gửi")
}

scope.launch {
    val msg = channel.receive()
    println("Đã nhận: $msg")
}

Ví dụ thứ hai — nhiều consumer trên một kênh duy nhất (fan-out):

val channel = Channel<Int>(Channel.UNLIMITED)

scope.launch {
    for (x in 1..10) channel.send(x)
    channel.close()
}

repeat(2) { id ->
    scope.launch {
        for (msg in channel) {
            println("Consumer #$id: $msg")
        }
    }
}

Ví dụ thứ ba — sử dụng builder produce với xử lý lỗi:

val source = produce {
    for (i in 1..5) {
        delay(200)
        send(i)
    }
}

scope.launch {
    source
        .consumeAsFlow()
        .catch { println("Lỗi: $it") }
        .collect { println("Phần tử: $it") }
}

Channel vs Flow

Channel là một nguyên hàm nóng: dữ liệu được phát ra độc lập với người đăng ký. Flow là lạnh: dữ liệu được tạo ra khi đăng ký. Channel hỗ trợ nhiều producer và consumer với phân phối đảm bảo mỗi phần tử đến một consumer (fan-out). Flow không được thiết kế cho nhiều producer độc lập.

Channel sử dụng bộ đệm với dung lượng có thể cấu hình và các hàm suspend send/receive để quản lý backpressure. Flow sử dụng cơ chế suspend collect với backpressure tự động qua coroutine. Channel là một công cụ cấp thấp cho các kịch bản cụ thể: chuyển đổi callback, mô hình actor, hàng đợi tác vụ với nhiều người gửi.

Đối với các kịch bản hàng ngày trong Android (trạng thái UI, luồng phản ứng từ DB), Google khuyến nghị Flow thay vì Channel. Channel nên được sử dụng khi cần trao đổi dữ liệu nóng giữa các coroutine với kiểm soát bộ đệm chính xác, hoặc khi chuyển đổi giao diện callback qua callbackFlow, mà triển khai nội bộ sử dụng Channel.

Một ví dụ thực tế quan trọng: khi triển khai client WebSocket, Channel cho phép ghi tin nhắn từ một coroutine và đọc từ coroutine khác với đảm bảo rằng mỗi tin nhắn sẽ được xử lý chính xác một lần. Flow không phù hợp cho nhiệm vụ này vì nó lạnh và không hỗ trợ nhiều producer. Channel với dung lượng UNLIMITED đảm bảo rằng các tin nhắn đến không bị mất trong thời gian trễ tạm thời của consumer.

Quản lý vòng đời kênh là một phần quan trọng khi làm việc với Channel. Kênh phải được đóng khi tất cả dữ liệu đã được gửi để consumer có thể hoàn thành vòng lặp. Gọi channel.close() báo hiệu rằng sẽ không có phần tử mới nào. Consumer có thể lặp qua for (item in channel) — vòng lặp sẽ tự động kết thúc sau close() và bộ đệm cạn kiệt. Thay vào đó, consumer có thể gọi receive() trong vòng lặp với xử lý ClosedReceiveChannelException.

Channel được sử dụng tích cực trong Android để triển khai EventBus không phụ thuộc: một Channel<Event> toàn cục với chiến lược Broadcast cho phép gửi sự kiện từ bất kỳ điểm nào trong ứng dụng. Không giống như các bus dựa trên LiveData, Channel không bị ràng buộc với vòng đời và không yêu cầu dọn dẹp khi chuyển đổi giữa các màn hình. send() từ ViewModel và receive() trong Activity/Fragment qua lifecycleScope cung cấp giao tiếp an toàn kiểu mà không cần lớp Event. Nhiều consumer trên một Channel phân phối tải — mỗi phần tử được xử lý một lần, ngăn xử lý trùng lặp một sự kiện duy nhất trong các người đăng ký khác nhau.

Trong hệ thống actor, Channel đóng vai trò nền tảng để triển khai mailbox — hàng đợi tin nhắn cho actor. Actor là một coroutine đọc tin nhắn từ Channel trong vòng lặp và xử lý chúng tuần tự. Cách tiếp cận này đảm bảo rằng mỗi tin nhắn được xử lý theo thứ tự gửi, không có điều kiện đua. Kotlin không có actor tích hợp như một kiểu (không giống như Akka), nhưng Channel + launch là một giải pháp thay thế nhẹ.

Để trao đổi hai chiều, cặp kênh được sử dụng: một kênh cho yêu cầu từ client đến server, kênh thứ hai cho phản hồi từ server đến client. Ví dụ, khi triển khai Pipe trong ứng dụng đa luồng: producer ghi vào OutputChannel, consumer đọc từ InputChannel. Các hàm suspend send và receive đảm bảo rằng Producer-Consumer sẽ không làm tràn ngăn xếp cuộc gọi, vì coroutine tạm dừng thay vì chặn. Channel với dung lượng BUFFERED phù hợp cho hầu hết các kịch bản nơi tốc độ producer và consumer xấp xỉ bằng nhau. Đối với các kịch bản bất đối xứng, hãy sử dụng UNLIMITED để producer không bị tạm dừng khi consumer bận — điều này giảm rủi ro deadlock nhưng tăng tiêu thụ bộ nhớ.

Chọn dung lượng Channel

Khi thiết kế kiến trúc với các kênh, điều quan trọng là nhớ về capacity: lựa chọn dung lượng ảnh hưởng trực tiếp đến hành vi dưới tải đỉnh. Các kênh với dung lượng BUFFERED(N) hoạt động như bộ đệm làm mịn: nếu consumer tạm thời chậm hơn producer, các phần tử tích tụ. Nếu tốc độ trung bình của consumer thấp hơn producer một cách nhất quán, bộ đệm sẽ đầy và coroutine gửi sẽ bị tạm dừng — đây là backpressure tự động bảo vệ khỏi quá tải bộ nhớ.

Để giám sát và gỡ lỗi Channel, hãy sử dụng kotlinx-coroutines-debug: tiện ích hiển thị số lượng coroutine đang hoạt động, trạng thái kênh của chúng (mở/đóng, số lượng phần tử trong bộ đệm) và ngăn xếp cuộc gọi của các thao tác send/receive bị tạm dừng. Channel cũng có thể được bọc trong một proxy ghi nhật ký: lớp LoggingChannel<T> ủy quyền cuộc gọi đến Channel thực, ghi nhật ký các thao tác send, receive và close. Điều này giúp xác định rò rỉ kênh khi close() không được gọi và coroutine consumer chờ vô hạn các phần tử mới.