Channel: چیست، انواع کانال‌ها و کوروتین‌ها در Kotlin

Channel در Kotlin چیست؟

Channel — از نظر مفهومی شبیه BlockingQueue در جاوا است، اما با توابع suspend send() و receive() به جای put() و take() مسدودکننده. توسعه‌دهنده Kotlin از Channel برای سازماندهی تبادل داده بین کوروتین‌ها بدون همگام‌سازی از طریق حافظه مشترک استفاده می‌کند. کانال تحویل مرتب را تضمین می‌کند — ترتیب ارسال با ترتیب دریافت مطابقت دارد.

ایجاد کانال

برای ایجاد Channel، تابع کارخانه‌ای Channel<T>(capacity) فراخوانی می‌شود. پارامتر capacity نوع کانال را تعیین می‌کند: RENDEZVOUS (0)، UNLIMITED (Int.MAX_VALUE)، CONFLATED (1-) یا یک عدد مشخص. نوع عنصر T توسط جنریک مشخص می‌شود. بستن کانال از طریق close() علامت می‌دهد که عناصر جدیدی وجود نخواهد داشت.

Send و Receive

send(value) — تابع suspend است که اگر کانال پر باشد، کوروتین فرستنده را متوقف می‌کند. receive() — تابع suspend است که اگر کانال خالی باشد، گیرنده را متوقف می‌کند. جایگزین trySend() و tryReceive() — نسخه‌های غیرمسدودکننده هستند که در صورت عدم امکان عملیات Boolean یا null برمی‌گردانند. آنها در زمینه‌های غیر suspend مفید هستند.

انواع Channel در kotlinx.coroutines

Kotlin چهار نوع Channel را از طریق ظرفیت بافر ارائه می‌دهد: Rendezvous (ظرفیت 0)، Buffered (ظرفیت N)، Conflated (ظرفیت 1، بازنویسی) و Unlimited (ظرفیت Int.MAX_VALUE). هر نوع وظیفه خود را حل می‌کند، از همگام‌سازی دقیق تا بافرینگ انبوه داده.

Rendezvous Channel — سخت‌گیرانه‌ترین: send() تا فراخوانی receive() در کوروتین دیگر مسدود می‌شود. در اصل، این یک نقطه rendezvous دو کوروتین است. برای handshake دقیق ایده‌آل است، زمانی که فرستنده باید صبر کند تا گیرنده عنصر را پردازش کند. از دست رفتن داده منتفی است — send تا زمانی که receive انجام نشود کامل نمی‌شود.

Conflated Channel — فقط آخرین مقدار ارسال شده را نگه می‌دارد. اگر فرستنده قبل از اینکه گیرنده مقدار قدیمی را بردارد، عنصر جدیدی قرار دهد، مقدار قدیمی کنار گذاشته می‌شود. Conflated Channel برای وضعیت UI مفید است: اگر کاربر به سرعت لغزنده را تغییر دهد، مقادیر میانی را می‌توان کنار گذاشت و فقط آخرین مقدار پردازش شود.

Producer-Consumer روی کانال‌ها

الگوی کلاسیک Producer-Consumer روی Channel از طریق کوروتین‌های موازی پیاده‌سازی می‌شود. Producer در حلقه send(value) را فراخوانی می‌کند، consumer — receive(value). تولیدکننده و مصرف‌کننده می‌توانند روی Dispatchers مختلف کار کنند: producer روی Dispatchers.IO، consumer روی Dispatchers.Main. Channel به طور خودکار دسترسی را بدون Lock و synchronized همگام‌سازی می‌کند.

Fan-out — چندین مصرف‌کننده روی یک کانال. هر عنصر دقیقاً به یک مصرف‌کننده می‌رسد (توزیع round-robin). Fan-in — چندین تولیدکننده در یک کانال می‌نویسند. کوروتین‌های فرستنده برای ارسال رقابت می‌کنند، اما ترتیب عناصر حفظ می‌شود. هر دو سناریو نیاز به همگام‌سازی اضافی ندارند.

Produce — یک سازنده کوروتین است که کانالی با بسته شدن خودکار ایجاد می‌کند. تابع produce { } یک ReceiveChannel برمی‌گرداند — کانال فقط خواندنی برای مصرف‌کننده. در داخل سازنده، send() داده ارسال می‌کند و پس از اتمام بلوک یا در صورت استثنا، کانال به طور خودکار بسته می‌شود و از نشت جلوگیری می‌کند.

Select و مالتی‌پلکس کردن

کتابخانه kotlinx.coroutines select را ارائه می‌دهد — عبارتی که منتظر اولین کانال تکمیل‌شده از چندین گزینه می‌ماند. Select اجازه می‌دهد چندین کانال را مالتی‌پلکس کنید: مثلاً منتظر داده از دو منبع بمانید و آن را که اول پاسخ داد پردازش کنید. نحو — select<T> { channel1.onReceive { } channel2.onReceive { } }. این جایگزینی برای عملگر amb در Rx است.

نمونه کد Channel

مثال اول — ساده‌ترین Rendezvous Channel، جایی که فرستنده منتظر دریافت است:

val channel = Channel<String>()

scope.launch {
    channel.send("Hello")
    println("ارسال شد")
}

scope.launch {
    val msg = channel.receive()
    println("دریافت شد: $msg")
}

مثال دوم — چندین مصرف‌کننده روی یک کانال (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("مصرف‌کننده #$id: $msg")
        }
    }
}

مثال سوم — استفاده از سازنده produce با مدیریت خطا:

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

scope.launch {
    source
        .consumeAsFlow()
        .catch { println("خطا: $it") }
        .collect { println("عنصر: $it") }
}

Channel در مقابل Flow

Channel — یک اولیه hot است: داده‌ها مستقل از مشترکین منتشر می‌شوند. Flow — cold: داده‌ها هنگام اشتراک تولید می‌شوند. Channel از چندین تولیدکننده و مصرف‌کننده با تضمین تحویل هر عنصر به یک مصرف‌کننده (fan-out) پشتیبانی می‌کند. Flow برای تولیدکنندگان مستقل متعدد طراحی نشده است.

Channel از بافر با ظرفیت قابل تنظیم و توابع suspend send/receive برای مدیریت backpressure استفاده می‌کند. Flow از مکانیزم suspend collect با backpressure خودکار از طریق کوروتین‌ها استفاده می‌کند. Channel — ابزاری سطح پایین برای سناریوهای خاص: تبدیل callback، مدل actor، صف وظایف با فرستندگان متعدد.

برای سناریوهای روزمره در Android (وضعیت UI، استریم‌های واکنشی از دیتابیس) Google Flow را توصیه می‌کند، نه Channel را. Channel باید زمانی استفاده شود که تبادل داده hot بین کوروتین‌ها با کنترل دقیق بافرینگ نیاز است، یا هنگام تبدیل رابط‌های callback از طریق callbackFlow که پیاده‌سازی داخلی آن از Channel استفاده می‌کند.

یک مثال عملی مهم: در پیاده‌سازی کلاینت WebSocket، Channel اجازه می‌دهد از یک کوروتین پیام بنویسید و از دیگری بخوانید با تضمین اینکه هر پیام دقیقاً یک بار پردازش می‌شود. Flow برای این کار مناسب نیست، زیرا cold است و از چندین تولیدکننده پشتیبانی نمی‌کند. Channel با ظرفیت UNLIMITED تضمین می‌کند که پیام‌های ورودی در تأخیرهای موقت مصرف‌کننده از دست نروند.

مدیریت چرخه حیات کانال — بخش مهم کار با Channel است. کانال باید بسته شود وقتی همه داده‌ها ارسال شده‌اند تا مصرف‌کننده بتواند تکرار را کامل کند. فراخوانی channel.close() علامت می‌دهد که عناصر جدیدی وجود نخواهد داشت. مصرف‌کننده می‌تواند از طریق for (item in channel) تکرار کند — حلقه پس از close() و خالی شدن بافر به طور خودکار پایان می‌یابد. به عنوان جایگزین، مصرف‌کننده می‌تواند receive() را در حلقه با مدیریت ClosedReceiveChannelException فراخوانی کند.

Channel به طور فعال در Android برای پیاده‌سازی EventBus بدون وابستگی استفاده می‌شود: Channel<Event> سراسری با استراتژی Broadcast اجازه می‌دهد رویدادها از هر نقطه برنامه ارسال شوند. برخلاف گذرگاه مبتنی بر LiveData، Channel به lifecycle وابسته نیست و هنگام انتقال بین صفحه‌ها نیاز به صفر کردن ندارد. send() از ViewModel و receive() در Activity/Fragment از طریق lifecycleScope ارتباط نوع-ایمن بدون کلاس‌های Event فراهم می‌کنند. چندین مصرف‌کننده روی Channel بار را توزیع می‌کنند — هر عنصر یک بار پردازش می‌شود که از تکراری شدن پردازش یک رویداد در مشترکین مختلف جلوگیری می‌کند.

در سیستم‌های actor، Channel به عنوان پایه‌ای برای پیاده‌سازی mailbox — صف پیام برای actor عمل می‌کند. Actor — کوروتینی است که در حلقه پیام‌ها را از Channel می‌خواند و آنها را به ترتیب پردازش می‌کند. این رویکرد تضمین می‌کند که هر پیام به ترتیب ارسال، بدون مسابقه داده پردازش می‌شود. Kotlin بر خلاف Akka، actor داخلی به عنوان نوع ندارد، اما Channel + launch جایگزین سبکی است.

برای تبادل دوطرفه از جفت کانال استفاده می‌شود: یک کانال برای درخواست‌ها از کلاینت به سرور، دومی — برای پاسخ‌ها از سرور به کلاینت. مثلاً در پیاده‌سازی Pipe در برنامه چندنخی: تولیدکننده در OutputChannel می‌نویسد، مصرف‌کننده از InputChannel می‌خواند. توابع suspend send و receive تضمین می‌کنند که Producer-Consumer پشته فراخوانی را سرریز نمی‌کند، زیرا کوروتین‌ها مسدود نمی‌شوند، بلکه متوقف می‌شوند. Channel با capacity BUFFERED برای اکثر سناریوهایی مناسب است که سرعت تولیدکننده و مصرف‌کننده تقریباً برابر است. برای سناریوهای نامتقارن از UNLIMITED استفاده کنید تا تولیدکننده وقتی مصرف‌کننده مشغول است متوقف نشود — این خطر deadlock را کاهش می‌دهد، اما مصرف حافظه را افزایش می‌دهد.

انتخاب ظرفیت Channel

هنگام طراحی معماری روی کانال‌ها مهم است که capacity را به خاطر بسپارید: انتخاب ظرفیت مستقیماً بر رفتار تحت بار پیک تأثیر می‌گذارد. کانال‌های با ظرفیت BUFFERED(N) مانند بافر صاف‌کننده عمل می‌کنند: اگر مصرف‌کننده موقتاً کندتر از تولیدکننده باشد، عناصر جمع می‌شوند. اگر سرعت متوسط مصرف‌کننده به طور پایدار کمتر از تولیدکننده باشد، بافر پر می‌شود و کوروتین فرستنده متوقف می‌شود — این backpressure خودکار است که از بارگذاری بیش از حد حافظه محافظت می‌کند.

برای مانیتورینگ و رفع اشکال Channel از kotlinx-coroutines-debug استفاده کنید: این ابزار تعداد کوروتین‌های فعال، وضعیت کانال‌های آنها (باز/بسته، تعداد عناصر در بافر) و پشته فراخوانی عملیات send/receive متوقف شده را نشان می‌دهد. Channel همچنین می‌تواند در یک پروکسی لاگ‌کننده پیچیده شود: کلاس LoggingChannel<T> فراخوانی‌ها را به Channel واقعی واگذار می‌کند و عملیات send، receive و close را ثبت می‌کند. این به شناسایی نشت کانال‌ها کمک می‌کند، زمانی که close() فراخوانی نشده و کوروتین مصرف‌کننده همیشه منتظر عناصر جدید است.