البرمجة التفاعلية في Go: مكتبة RxGo (Reactive Extensions for Go)

البرمجة التفاعلية في Go (Golang) مع مكتبة RxGo

البرمجة التفاعلية هي نموذج برمجي يركز على تدفقات البيانات غير المتزامنة ونشر التغييرات. في نظام Go (Golang) البيئي، تُعد مكتبة RxGo (Reactive Extensions for Go) واحدة من أقوى تطبيقات هذا النموذج. تتيح هذه المكتبة للمطورين العمل مع تسلسلات الأحداث باستخدام أسلوب تصريحي، وهو مفيد بشكل خاص عند معالجة إدخال المستخدم، وطلبات الشبكة، والرسائل من القنوات، وأي مصادر بيانات غير متزامنة أخرى.

ما هي RxGo؟

RxGo هي نسخة من مكتبة ReactiveX (Reactive Extensions) الشهيرة للغة Go. الفكرة الأساسية هي إنشاء Observable (تدفق قابل للملاحظة) يُصدر عناصر، وObserver (مراقب) يُشترك في هذا التدفق ويستجيب للبيانات الواردة أو الأخطاء أو إكمال التدفق.

توفر المكتبة مجموعة غنية من العوامل (Operators) لتحويل التدفقات وتصفيتها ودمجها والتحكم فيها. الميزة الرئيسية هي أن الكود يصبح موجزًا وقابلاً للقراءة وسهل الاختبار، خاصة في السيناريوهات المعقدة التي تتضمن مصادر متعددة للأحداث.

تثبيت RxGo

للبدء، قم بتثبيت الحزمة عبر go get:

go get -u github.com/reactivex/rxgo/v2

تأكد من استخدام الإصدار v2، حيث أنه يحتوي على واجهة برمجة التطبيقات (API) الحالية ودعم سياقات Go.

المفاهيم الأساسية والأمثلة

1. إنشاء Observable

أبسط طريقة لإنشاء تدفق هي استخدام rxgo.Just أو rxgo.FromChannel. لنلق نظرة على مثال باستخدام Just:

package main\n\nimport (\n    "context"\n    "fmt"\n    "github.com/reactivex/rxgo/v2"\n)\n\nfunc main() {\n    observable := rxgo.Just(1, 2, 3, 4, 5)()\n    ch := observable.Observe()\n    for item := range ch {\n        if item.Error() {\n            fmt.Println("خطأ:", item.Err)\n        } else {\n            fmt.Println("تم الاستلام:", item.V)\n        }\n    }\n}

يقوم هذا الكود بإنشاء تدفق من خمسة أرقام. تُرجع طريقة Observe() قناة يتم من خلالها استلام العناصر. هام: في RxGo يتم استخدام backpressure (الضغط العكسي) - إذا كان المستهلك بطيئًا، يتم تخزين البيانات مؤقتًا.

2. عوامل التحويل (Transformation Operators)

العوامل هي قلب البرمجة التفاعلية. على سبيل المثال، يقوم Map بتحويل كل عنصر:

package main\n\nimport (\n    "context"\n    "fmt"\n    "github.com/reactivex/rxgo/v2"\n)\n\nfunc main() {\n    observable := rxgo.Just(1, 2, 3)().\n        Map(func(ctx context.Context, i interface{}) (interface{}, error) {\n            num := i.(int)\n            return num * 10, nil\n        }, rxgo.WithPool(2)) // معالجة متوازية\n\n    for item := range observable.Observe() {\n        fmt.Println(item.V)\n    }\n}

المخرجات: 10, 20, 30. يقوم العامل Map بتطبيق دالة على كل عنصر. المعامل rxgo.WithPool(2) يُفعّل المعالجة المتوازية باستخدام اثنتين من goroutines.

3. تصفية ودمج التدفقات

باستخدام Filter يمكن تصفية العناصر غير المرغوب فيها، بينما يقوم Merge بدمج تدفقات متعددة:

package main\n\nimport (\n    "fmt"\n    "github.com/reactivex/rxgo/v2"\n)\n\nfunc main() {\n    // تصفية الأرقام الزوجية\n    even := rxgo.Just(1, 2, 3, 4, 5)().\n        Filter(func(i interface{}) bool {\n            return i.(int)%2 == 0\n        })\n\n    // تدفق ثانٍ\n    odd := rxgo.Just(10, 20, 30)()\n\n    // الدمج\n    merged := rxgo.Merge(even, odd)\n\n    for item := range merged.Observe() {\n        fmt.Println(item.V) // 2, 4, 10, 20, 30\n    }\n}

4. العمل مع القنوات والسياقات

تتكامل RxGo بشكل ممتاز مع قنوات Go الأصلية. لنقم بإنشاء Observable من قناة وإضافة