Channel: ما هو، أنواع القنوات والكوروتينات في Kotlin

ما هو Channel في Kotlin؟

Channel مشابه conceptually لـ BlockingQueue من Java، لكن مع دوال 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() في كوروتين آخر. في جوهره، هو نقطة التقاء لكوروتينين. مثالي للمصافحة الصارمة عندما يجب على المرسل انتظار معالجة المستلم للعنصر. يتم استبعاد فقدان البيانات — لا يكتمل send حتى يتم تنفيذ receive.

Conflated Channel يخزن فقط آخر قيمة تم إرسالها. إذا وضع المرسل عنصرًا جديدًا قبل أن يلتقط المستلم العنصر القديم، يتم التخلص من القديم. Conflated Channel مفيد لحالة واجهة المستخدم: إذا قام المستخدم بتغيير شريط التمرير بسرعة، يمكن التخلص من القيم الوسيطة ومعالجة القيمة الأخيرة فقط.

منتج-مستهلك مع القنوات

يتم تنفيذ نمط المنتج-المستهلك الكلاسيكي على Channel من خلال كوروتينات متوازية. المنتج يستدعي send(value) في حلقة، والمستهلك يستدعي receive(value). يمكن للمنتج والمستهلك العمل على Dispatchers مختلفة: المنتج على Dispatchers.IO، والمستهلك على 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("Consumer #$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 هو بدائي ساخن: يتم إصدار البيانات بشكل مستقل عن المشتركين. Flow بارد: يتم إنشاء البيانات عند الاشتراك. يدعم Channel منتجين ومستهلكين متعددين مع تسليم مضمون لكل عنصر إلى مستهلك واحد (fan-out). Flow غير مصمم لمنتجين مستقلين متعددين.

يستخدم Channel مخزنًا مؤقتًا بسعة قابلة للتكوين ودوال suspend send/receive لإدارة الضغط العكسي. يستخدم Flow آلية suspend collect مع ضغط عكسي تلقائي عبر الكوروتينات. Channel هو أداة منخفضة المستوى لسيناريوهات محددة: تحويل callback، نموذج الممثل، قائمة مهام مع مرسلين متعددين.

للسيناريوهات اليومية في Android (حالة واجهة المستخدم، التدفقات التفاعلية من قاعدة البيانات) توصي Google بـ Flow بدلاً من Channel. يجب استخدام Channel عندما يكون هناك حاجة لتبادل بيانات ساخن بين الكوروتينات مع تحكم دقيق في المخزن المؤقت، أو عند تحويل واجهات callback عبر callbackFlow، الذي يستخدم تنفيذه الداخلي Channel.

مثال عملي مهم: عند تنفيذ عميل WebSocket، يسمح Channel بكتابة الرسائل من كوروتين واحد وقراءتها من آخر مع ضمان معالجة كل رسالة مرة واحدة بالضبط. Flow غير مناسب لهذه المهمة لأنه بارد ولا يدعم منتجين متعددين. Channel بسعة UNLIMITED يضمن عدم فقدان الرسائل الواردة أثناء تأخيرات المستهلك المؤقتة.

إدارة دورة حياة القناة هي جزء مهم من العمل مع Channel. يجب إغلاق القناة عندما يتم إرسال جميع البيانات حتى يتمكن المستهلك من إنهاء التكرار. استدعاء channel.close() يشير إلى أنه لن يكون هناك عناصر جديدة. يمكن للمستهلك التكرار عبر for (item in channel) — ستنتهي الحلقة تلقائيًا بعد close() واستنزاف المخزن المؤقت. بدلاً من ذلك، يمكن للمستهلك استدعاء receive() في حلقة مع معالجة ClosedReceiveChannelException.

يستخدم Channel بنشاط في Android لتنفيذ EventBus بدون تبعيات: Channel<Event> عالمي مع استراتيجية Broadcast يسمح بإرسال الأحداث من أي نقطة في التطبيق. على عكس الحافلات القائمة على LiveData، فإن Channel غير مرتبط بدورة الحياة ولا يتطلب المسح عند الانتقال بين الشاشات. send() من ViewModel و receive() في Activity/Fragment عبر lifecycleScope يوفران اتصالاً آمنًا من حيث الأنواع بدون فئات Event. يقوم المستهلكون المتعددون على Channel بتوزيع الحمل — تتم معالجة كل عنصر مرة واحدة، مما يمنع المعالجة المكررة لحدث واحد في مشتركين مختلفين.

في أنظمة الممثلين، يعمل Channel كأساس لتنفيذ mailbox — قائمة انتظار الرسائل للممثل. الممثل هو كوروتين يقرأ الرسائل من Channel في حلقة ويعالجها بالتسلسل. يضمن هذا النهج معالجة كل رسالة بترتيب الإرسال، دون سباقات البيانات. ليس لدى Kotlin ممثل مدمج كنوع (على عكس Akka)، لكن Channel + launch هو بديل خفيف.

للتبادل ثنائي الاتجاه، يتم استخدام أزواج القنوات: قناة للطلبات من العميل إلى الخادم، والثانية للردود من الخادم إلى العميل. على سبيل المثال، عند تنفيذ Pipe في تطبيق متعدد الخيوط: يكتب المنتج في OutputChannel، ويقرأ المستهلك من InputChannel. تضمن دالتا suspend send و receive أن المنتج-المستهلك لن يفيض مكدس الاستدعاءات، حيث يتم تعليق الكوروتينات بدلاً من حظرها. Channel بسعة BUFFERED مناسب لمعظم السيناريوهات حيث تكون سرعة المنتج والمستهلك متساوية تقريبًا. للسيناريوهات غير المتماثلة، استخدم UNLIMITED حتى لا يتم تعليق المنتج عندما يكون المستهلك مشغولاً — هذا يقلل من خطر الجمود لكنه يزيد من استهلاك الذاكرة.

اختيار سعة Channel

عند تصميم بنية باستخدام القنوات، من المهم تذكر capacity: اختيار السعة يؤثر مباشرة على السلوك تحت الحمل الأقصى. القنوات بسعة BUFFERED(N) تعمل كمخزن مؤقت ممهد: إذا كان المستهلك أبطأ مؤقتًا من المنتج، تتراكم العناصر. إذا كانت سرعة المستهلك المتوسطة أقل باستمرار من المنتج، سيمتلئ المخزن المؤقت وسيتم تعليق الكوروتين المرسل — هذا هو الضغط العكسي التلقائي الذي يحمي من زيادة تحميل الذاكرة.

لمراقبة وتصحيح أخطاء Channel استخدم kotlinx-coroutines-debug: تعرض الأداة عدد الكوروتينات النشطة، حالة قنواتها (مفتوح/مغلق، عدد العناصر في المخزن المؤقت) ومكدس استدعاءات عمليات send/receive المعلقة. يمكن أيضًا تغليف Channel في وكيل تسجيل: الفئة LoggingChannel<T> تفوض الاستدعاءات إلى Channel الحقيقي، مسجلة عمليات send و receive و close. يساعد هذا في تحديد تسرب القنوات عندما لا يتم استدعاء close() وتنتظر كوروتين المستهلك إلى الأبد عناصر جديدة.