Channel: nedir, kanal türleri ve Kotlin'de coroutine'ler

Kotlin'de Channel Nedir?

Channel kavramsal olarak Java'daki BlockingQueue'ye benzer, ancak bloke edici put() ve take() yerine suspend fonksiyonlar send() ve receive() ile çalışır. Bir Kotlin geliştiricisi, paylaşılan bellek üzerinden senkronizasyon olmadan coroutine'ler arasında veri alışverişini düzenlemek için Channel'ı kullanır. Kanal sıralı teslimatı garanti eder — gönderme sırası, alma sırasıyla eşleşir.

Kanal oluşturma

Channel oluşturmak için fabrika fonksiyonu Channel<T>(capacity) çağrılır. capacity parametresi kanal türünü belirler: RENDEZVOUS (0), UNLIMITED (Int.MAX_VALUE), CONFLATED (-1) veya belirli bir sayı. Eleman türü T, jenerikler aracılığıyla belirtilir. close() ile kanalı kapatmak, yeni eleman gelmeyeceğini bildirir.

Send ve Receive

send(value), kanal doluysa gönderen coroutine'i askıya alan bir suspend fonksiyonudur. receive(), kanal boşsa alıcıyı askıya alan bir suspend fonksiyonudur. Alternatifler trySend() ve tryReceive(), işlem mümkün olmadığında Boolean veya null döndüren bloke etmeyen sürümlerdir. Suspend olmayan bağlamlarda kullanışlıdırlar.

kotlinx.coroutines'ta Channel Türleri

Kotlin, arabellek kapasitesi aracılığıyla dört Channel varyantı sağlar: Rendezvous (kapasite 0), Buffered (kapasite N), Conflated (kapasite 1, üzerine yazma) ve Unlimited (kapasite Int.MAX_VALUE). Her tür, katı senkronizasyondan toplu veri arabelleklemesine kadar kendi görevini çözer.

Rendezvous Channel en katı olanıdır: başka bir coroutine'de receive() çağrılana kadar send() bloke olur. Özünde, iki coroutine'in buluşma noktasıdır. Gönderenin, alıcının elemanı işlemesini beklemesi gereken katı el sıkışma için idealdir. Veri kaybı hariç tutulur — receive yürütülene kadar send tamamlanmaz.

Conflated Channel yalnızca gönderilen son değeri saklar. Gönderen, alıcı eski değeri almadan önce yeni bir eleman koyarsa, eski değer atılır. Conflated Channel, kullanıcı arayüzü durumu için kullanışlıdır: bir kullanıcı kaydırıcıyı hızlıca değiştirirse, ara değerler atılabilir ve yalnızca son değer işlenebilir.

Kanallarla Üretici-Tüketici

Channel üzerindeki klasik Üretici-Tüketici deseni, paralel coroutine'ler aracılığıyla uygulanır. Üretici bir döngüde send(value) çağırır, tüketici receive(value) çağırır. Üretici ve tüketici farklı Dispatcher'larda çalışabilir: üretici Dispatchers.IO'da, tüketici Dispatchers.Main'de. Channel, Lock veya synchronized olmadan erişimi otomatik olarak senkronize eder.

Fan-out — tek bir kanalda birden çok tüketici. Her eleman tam olarak bir tüketiciye gider (round-robin dağıtımı). Fan-in — birden çok üretici tek bir kanala yazar. Gönderen coroutine'ler gönderme için rekabet eder, ancak elemanların sırası korunur. Her iki senaryo da ek senkronizasyon gerektirmez.

Produce, otomatik kapanma ile bir kanal oluşturan bir coroutine builder'dır. produce { } fonksiyonu, tüketici için salt okunur bir kanal olan ReceiveChannel döndürür. Builder içinde send() veri gönderir ve blok tamamlandığında veya bir istisna oluştuğunda kanal otomatik olarak kapanarak sızıntıları önler.

Select ve çoğullama

kotlinx.coroutines kütüphanesi select sağlar — birden çok alternatif arasından ilk tamamlanan kanalı bekleyen bir ifade. Select, birden çok kanalı çoğullamaya izin verir: örneğin, iki kaynaktan veri beklemek ve ilk yanıt vereni işlemek. Sözdizimi — select<T> { channel1.onReceive { } channel2.onReceive { } }. Bu, Rx'teki amb operatörüne bir alternatiftir.

Channel Kod Örnekleri

İlk örnek, gönderenin almayı beklediği basit bir Rendezvous Channel'dır:

val channel = Channel<String>()

scope.launch {
    channel.send("Hello")
    println("Gönderildi")
}

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

İkinci örnek — tek bir kanalda birden çok tüketici (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")
        }
    }
}

Üçüncü örnek — hata yönetimiyle produce builder kullanımı:

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

scope.launch {
    source
        .consumeAsFlow()
        .catch { println("Hata: $it") }
        .collect { println("Eleman: $it") }
}

Channel vs Flow

Channel sıcak bir ilkeldir: veriler abonelerden bağımsız olarak yayılır. Flow soğuktur: veriler abonelikte oluşturulur. Channel, her elemanın bir tüketiciye garantili teslimatı (fan-out) ile birden çok üretici ve tüketiciyi destekler. Flow, birden çok bağımsız üretici için tasarlanmamıştır.

Channel, geri basınç yönetimi için yapılandırılabilir kapasiteli bir arabellek ve suspend fonksiyonlar send/receive kullanır. Flow, coroutine'ler aracılığıyla otomatik geri basınç ile suspend mekanizması collect'i kullanır. Channel, belirli senaryolar için düşük seviyeli bir araçtır: geri arama dönüşümü, aktör modeli, birden çok gönderenli görev kuyruğu.

Android'deki günlük senaryolar (kullanıcı arayüzü durumu, veritabanından reaktif akışlar) için Google, Channel yerine Flow önerir. Channel, hassas arabellek kontrolü ile coroutine'ler arasında sıcak veri alışverişi gerektiğinde veya iç uygulaması Channel kullanan callbackFlow aracılığıyla geri arama arayüzleri dönüştürülürken kullanılmalıdır.

Önemli bir pratik örnek: bir WebSocket istemcisi uygulanırken, Channel bir coroutine'den mesaj yazmaya ve diğerinden okumaya izin verir ve her mesajın tam olarak bir kez işleneceğini garanti eder. Flow bu görev için uygun değildir çünkü soğuktur ve birden çok üreticiyi desteklemez. UNLIMITED kapasiteli Channel, geçici tüketici gecikmeleri sırasında gelen mesajların kaybolmamasını sağlar.

Kanal yaşam döngüsü yönetimi, Channel ile çalışmanın önemli bir parçasıdır. Tüm veriler gönderildiğinde kanal kapatılmalıdır, böylece tüketici yinelemeyi tamamlayabilir. channel.close() çağırmak, yeni eleman gelmeyeceğini bildirir. Tüketici for (item in channel) ile yineleme yapabilir — döngü, close() ve ara belleğin boşaltılmasından sonra otomatik olarak sona erer. Alternatif olarak, tüketici ClosedReceiveChannelException yönetimi ile bir döngüde receive() çağırabilir.

Channel, Android'de bağımlılık olmadan EventBus uygulamak için aktif olarak kullanılır: Yayın stratejisine sahip global bir Channel<Event>, uygulamanın herhangi bir noktasından olay göndermeye izin verir. LiveData tabanlı veri yollarının aksine, Channel yaşam döngüsüne bağlı değildir ve ekranlar arasında geçiş yaparken temizleme gerektirmez. ViewModel'den send() ve Activity/Fragment'te lifecycleScope aracılığıyla receive(), Event sınıfları olmadan tür güvenli iletişim sağlar. Bir Channel'daki birden çok tüketici yükü dağıtır — her eleman bir kez işlenir, bu da farklı abonelerde tek bir olayın yinelenen işlenmesini önler.

Aktör sistemlerinde Channel, bir posta kutusu — aktör için mesaj kuyruğu — uygulamanın temeli olarak hizmet eder. Bir aktör, bir döngüde Channel'dan mesajları okuyan ve bunları sırayla işleyen bir coroutine'dir. Bu yaklaşım, her mesajın veri yarışı olmadan gönderme sırasında işlenmesini garanti eder. Kotlin'de (Akka'nın aksine) yerleşik bir aktör türü yoktur, ancak Channel + launch hafif bir alternatiftir.

Çift yönlü değişim için kanal çiftleri kullanılır: biri istemciden sunucuya istekler için, ikincisi sunucudan istemciye yanıtlar için. Örneğin, çok iş parçacıklı bir uygulamada Pipe uygulanırken: üretici OutputChannel'a yazar, tüketici InputChannel'dan okur. Suspend fonksiyonları send ve receive, coroutine'ler bloke olmak yerine askıya alındığı için Üretici-Tüketici'nin çağrı yığınını taşırmamasını garanti eder. BUFFERED kapasiteli Channel, üretici ve tüketici hızının yaklaşık olarak eşit olduğu çoğu senaryo için uygundur. Asimetrik senaryolar için, tüketici meşgulken üreticinin askıya alınmaması için UNLIMITED kullanın — bu, kilitlenme riskini azaltır ancak bellek tüketimini artırır.

Channel Kapasitesi Seçimi

Kanallarla bir mimari tasarlarken capacity'yi hatırlamak önemlidir: kapasite seçimi, tepe yükü altındaki davranışı doğrudan etkiler. BUFFERED(N) kapasiteli kanallar, yumuşatma arabelleği gibi davranır: tüketici geçici olarak üreticiden daha yavaşsa, elemanlar birikir. Tüketicinin ortalama hızı sürekli olarak üreticiden düşükse, arabellek dolacak ve gönderen coroutine askıya alınacaktır — bu, bellek aşırı yüklenmesine karşı koruyan otomatik geri basınçtır.

Channel izleme ve hata ayıklama için kotlinx-coroutines-debug kullanın: yardımcı program, etkin coroutine'lerin sayısını, kanal durumlarını (açık/kapalı, arabellekteki eleman sayısı) ve askıya alınmış send/receive işlemlerinin çağrı yığınını gösterir. Channel ayrıca bir günlükleme proxy'sine sarılabilir: LoggingChannel<T> sınıfı, gerçek Channel'a çağrıları devrederek send, receive ve close işlemlerini günlükler. Bu, close() çağrılmadığında ve tüketici coroutine'inin sonsuza kadar yeni eleman beklediği kanal sızıntılarını belirlemeye yardımcı olur.