Channel: nədir, kanal növləri və Kotlin-də korutinlər

Kotlin-də Channel nədir?

Channel — konseptual olaraq Java-dakı BlockingQueue-ya bənzəyir, lakin bloklayan put() və take() əvəzinə suspend-funksiyalar send() və receive() istifadə edir. Kotlin tərtibatçısı Channel-i ümumi yaddaş vasitəsilə sinxronizasiya olmadan korutinlər arasında məlumat mübadiləsi təşkil etmək üçün istifadə edir. Kanal sifarişli çatdırılmaya zəmanət verir — göndərmə sırası qəbul sırası ilə üst-üstə düşür.

Kanalın yaradılması

Channel yaratmaq üçün fabrik funksiyası Channel<T>(capacity) çağırılır. capacity parametri kanalın növünü müəyyənləşdirir: RENDEZVOUS (0), UNLIMITED (Int.MAX_VALUE), CONFLATED (-1) və ya müəyyən rəqəm. T elementi növü generik vasitəsilə təyin edilir. Kanalın close() vasitəsilə bağlanması yeni elementlərin olmayacağını bildirir.

Send və Receive

send(value) — suspend-funksiyadır, kanal doludursa göndərən korutinanı dayandırır. receive() — suspend-funksiyadır, kanal boşdursa qəbul edəni dayandırır. Alternativ trySend() və tryReceive() — bloklanmayan versiyalardır, əməliyyat mümkün olmadıqda Boolean və ya null qaytarır. Onlar suspend olmayan kontekstlərdə faydalıdır.

kotlinx.coroutines-də Channel növləri

Kotlin bufer tutumu vasitəsilə dörd Channel variantı təqdim edir: Rendezvous (tutum 0), Buffered (tutum N), Conflated (tutum 1, üstə yazma) və Unlimited (tutum Int.MAX_VALUE). Hər növ öz vəzifəsini həll edir, ciddi sinxronizasiyadan kütləvi məlumat buferlənməsinə qədər.

Rendezvous Channel — ən ciddi: send() başqa korutinada receive() çağırılana qədər bloklanır. Əslində, bu iki korutinanın görüş nöqtəsidir. Ciddi handshake üçün idealdır, göndərən elementin qəbul edən tərəfindən işləndiyini gözləməlidir. Məlumat itkisi istisna edilir — send receive yerinə yetirilənə qədər tamamlanmaz.

Conflated Channel — yalnız son göndərilmiş dəyəri saxlayır. Əgər göndərən qəbul edən köhnəni götürməmişdən yeni element qoyubsa, köhnəsi atılır. Conflated Channel UI vəziyyəti üçün faydalıdır: istifadəçi sürətlə sürüşdürəni dəyişirsə, aralıq dəyərlər atıla bilər və yalnız sonuncu işlənir.

Kanallarda Producer-Consumer

Channel-da klassik Producer-Consumer nümunəsi paralel korutinlər vasitəsilə həyata keçirilir. Producer dövrədə send(value) çağırır, consumer — receive(value). İstehsalçı və istehlakçı müxtəlif Dispatchers-də işləyə bilər: producer Dispatchers.IO-da, consumer Dispatchers.Main-də. Channel Lock və synchronized olmadan avtomatik olaraq girişi sinxronlaşdırır.

Fan-out — bir kanalda çoxlu istehlakçı. Hər element dəqiq bir istehlakçıya çatır (round-robin paylanması). Fan-in — çoxlu istehsalçı bir kanala yazır. Göndərən korutinlər göndərmə üçün rəqabət aparır, lakin elementlərin sırası qorunur. Hər iki ssenari əlavə sinxronizasiya tələb etmir.

Produce — avtomatik bağlanma ilə kanal yaradan korutin qurucusudur. produce { } funksiyası ReceiveChannel qaytarır — istehlakçı üçün yalnız oxunan kanal. Qurucu daxilində send() məlumat göndərir və blok tamamlandıqda və ya istisna zamanı kanal avtomatik bağlanır, sızıntıların qarşısını alır.

Select və multipleksasiya

kotlinx.coroutines kitabxanası select — bir neçə alternativdən ilk tamamlanan kanalı gözləyən ifadəni təqdim edir. Select birdən çox kanalı multipleksasiya etməyə imkan verir: məsələn, iki mənbədən məlumat gözləmək və ilk cavab verəni emal etmək. Sintaksis — select<T> { channel1.onReceive { } channel2.onReceive { } }. Bu Rx-dəki amb operatoruna alternativdir.

Channel kod nümunələri

Birinci nümunə — ən sadə Rendezvous Channel, göndərənin qəbulu gözlədiyi:

val channel = Channel<String>()

scope.launch {
    channel.send("Hello")
    println("Göndərildi")
}

scope.launch {
    val msg = channel.receive()
    println("Alındı: $msg")
}

İkinci nümunə — bir kanalda bir neçə istehlakçı (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("İstehlakçı #$id: $msg")
        }
    }
}

Üçüncü nümunə — səhv idarəetmə ilə produce qurucusunun istifadəsi:

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

scope.launch {
    source
        .consumeAsFlow()
        .catch { println("Xəta: $it") }
        .collect { println("Element: $it") }
}

Channel vs Flow

Channel — isti primitivdir: məlumatlar abunəçilərdən asılı olmayaraq yayılır. Flow — soyuq: məlumatlar abunə zamanı yaradılır. Channel çoxlu istehsalçı və istehlakçını hər elementin bir istehlakçıya çatdırılması zəmanəti ilə dəstəkləyir (fan-out). Flow müstəqil istehsalçılar üçün nəzərdə tutulmayıb.

Channel backpressure idarəsi üçün konfiqurasiya olunan bufer tutumu və suspend-funksiyalar send/receive istifadə edir. Flow korutinlər vasitəsilə avtomatik backpressure ilə suspend mexanizmi collect istifadə edir. Channel — xüsusi ssenarilər üçün aşağı səviyyəli alətdir: callback çevrilməsi, aktyor modeli, çoxlu göndərən ilə tapşırıq növbəsi.

Android-də gündəlik ssenarilər üçün (UI vəziyyəti, DB-dən reaktiv axınlar) Google Flow-u tövsiyə edir, Channel-i yox. Channel korutinlər arasında dəqiq bufer nəzarəti ilə isti məlumat mübadiləsi lazım olduqda və ya callback interfeyslərinin callbackFlow vasitəsilə çevrilməsində istifadə edilməlidir, onun daxili tətbiqi Channel istifadə edir.

Əhəmiyyətli praktik nümunə: WebSocket müştərisinin tətbiqində, Channel bir korutindən mesaj yazmağa və digərindən oxumağa imkan verir ki, hər mesaj dəqiq bir dəfə emal edilsin. Flow bu tapşırıq üçün uyğun deyil, çünki soyuqdur və çoxlu istehsalçını dəstəkləmir. UNLIMITED tutumlu Channel daxil olan mesajların istehlakçının müvəqqəti gecikmələri zamanı itirilməməsini təmin edir.

Kanalın həyat dövranının idarə edilməsi — Channel ilə işin vacib hissəsidir. Bütün məlumatlar göndərildikdən sonra kanal bağlanmalıdır ki, istehlakçı iterasiyanı tamamlaya bilsin. channel.close() çağırışı yeni elementlərin olmayacağını bildirir. İstehlakçı for (item in channel) vasitəsilə iterasiya edə bilər — dövr close() və buferin boşaldılmasından sonra avtomatik bitir. Alternativ olaraq istehlakçı ClosedReceiveChannelException idarəetməsi ilə dövrədə receive() çağıra bilər.

Channel Android-də asılılıqsız EventBus tətbiqi üçün aktiv istifadə olunur: Broadcast strategiyası ilə qlobal Channel<Event> tətbiqin istənilən nöqtəsindən hadisələr göndərməyə imkan verir. LiveData üzərindəki şinadan fərqli olaraq, Channel lifecycle-a bağlı deyil və ekranlar arasında keçid zamanı sıfırlama tələb etmir. ViewModel-dən send() və Activity/Fragment-də lifecycleScope vasitəsilə receive() Event sinifləri olmadan tip təhlükəsiz ünsiyyət təmin edir. Channel-da çoxlu istehlakçı yükü paylayır — hər element bir dəfə emal olunur, bu da müxtəlif abunəçilərdə bir hadisənin emalının təkrarlanmasının qarşısını alır.

Aktyor sistemlərində Channel aktyor üçün mesaj növbəsi olan mailbox-un tətbiqi üçün əsas rolunu oynayır. Aktyor — dövrədə Channel-dan mesajları oxuyan və onları ardıcıl emal edən korutinadır. Bu yanaşma hər mesajın göndərmə sırası ilə, məlumat yarışları olmadan emal edilməsini təmin edir. Kotlin-də Akka-dan fərqli olaraq daxili aktyor növü yoxdur, lakin Channel + launch yüngül əvəzedicidir.

İkitərəfli mübadilə üçün kanal cütləri istifadə olunur: bir kanal müştəridən serverə sorğular üçün, digəri — serverdən müştəriyə cavablar üçün. Məsələn, çoxiplikli tətbiqidə Pipe tətbiqində: istehsalçı OutputChannel-a yazır, istehlakçı InputChannel-dan oxuyur. suspend-funksiyalar send və receive zəmanət verir ki, Producer-Consumer çağırış yığınını daşırmaz, çünki korutinlər bloklanmır, dayandırılır. BUFFERED tutumlu Channel istehsalçı və istehlakçının sürətlərinin təxminən bərabər olduğu əksər ssenarilər üçün uyğundur. Asimmetrik ssenarilər üçün UNLIMITED istifadə edin ki, istehsalçı istehlakçı məşğul olduqda dayandırılmasın — bu deadlock riskini azaldır, lakin yaddaş istehlakını artırır.

Channel tutumunun seçilməsi

Kanallarda arxitektura layihələndirilərkən capacity haqqında xatırlamaq vacibdir: tutumun seçimi pik yük altında davranışa birbaşa təsir edir. BUFFERED(N) tutumu olan kanallar hamarlaşdırıcı bufer kimi işləyir: əgər istehlakçı müvəqqəti olaraq istehsalçıdan yavaşdırsa, elementlər toplanır. Əgər istehlakçının orta sürəti istehsalçıdan sabit şəkildə aşağıdırsa, bufer dolacaq və sender korutinası dayandırılacaq — bu, yaddaşın həddən artıq yüklənməsindən qoruyan avtomatik backpressure-dır.

Channel-ın monitorinqi və debug edilməsi üçün kotlinx-coroutines-debug istifadə edin: alət aktiv korutinlərin sayını, onların kanallarının vəziyyətini (açıq/bağlı, buferdəki elementlərin sayı) və dayandırılmış send/receive əməliyyatlarının çağırış yığınını göstərir. Channel həmçinin qeydiyyatçı proxy ilə bükülə bilər: LoggingChannel<T> sinfi çağırışları real Channel-ə delegasiya edir, send, receive və close əməliyyatlarını qeydə alır. Bu, close() çağırılmadıqda və istehlakçı korutinasının sonsuz yeni elementlər gözlədikdə kanal sızıntılarını aşkar etməyə kömək edir.