Event Bus ونمط Observer في Go (Golang): تنفيذ pub/sub

Event Bus ونمط Observer في Go (Golang): من النظرية إلى التطبيق

نمط Observer (المراقب) وتنفيذه في شكل Event Bus (ناقل الأحداث) هما أدوات أساسية لبناء أنظمة مترابطة بشكل فضفاض وقابلة للتوسع. في Go، بفضل فلسفتها "تواصل عبر القنوات، وليس عبر الذاكرة المشتركة"، تكتسب هذه الأنماط أناقة خاصة. في هذه المقالة، سنشرح كيفية إنشاء Event Bus خاص بك في Go باستخدام goroutines والقنوات، وسنستعرض سيناريوهات الاستخدام النموذجية.

نمط Observer: الجوهر والمشكلة

تخيل أن لديك نظامًا حيث يجب على كائن واحد (الناشر) إخطار العديد من الكائنات الأخرى (المشتركين) بتغيير حالته. بدون Observer، ستحتاج إلى ربط الناشر بكل مشترك بشكل صارم. يحل Observer هذه المشكلة من خلال تقديم واجهة اشتراك: يحتفظ الناشر بقائمة المشتركين ويستدعي طريقة Update() الخاصة بهم عند وقوع الحدث.

Event Bus هو ناقل مركزي يعمل كوسيط. يمكن لأي مكون نشر حدث، ويمكن لأي مكون آخر الاشتراك فيه، دون معرفة وجود بعضهم البعض.

لماذا Go مثالية لـ Event Bus؟

\n
• Goroutines والقنوات: معالجة الأحداث غير المتزامنة دون حظر.
\n
• الواجهات: تسمح بتحديد العقود للمعالجات بسهولة.
\n
• غياب الوراثة: التركيب عبر القنوات والدوال أكثر طبيعية من التنفيذ الكلاسيكي للبرمجة كائنية التوجه.
\n

تنفيذ بسيط لـ Event Bus في Go

لنبدأ بإصدار أساسي حيث الأحداث هي نصوص والمشتركون هم دوال معالجة.

package main\n\nimport (\n    "fmt"\n    "sync"\n    "time"\n)\n\n// EventBus — ناقل الأحداث الخاص بنا\ntype EventBus struct {\n    subscribers map[string][]func(string)\n    mu          sync.RWMutex\n}\n\n// NewEventBus ينشئ ناقل أحداث جديد\nfunc NewEventBus() *EventBus {\n    return &EventBus{\n        subscribers: make(map[string][]func(string)),\n    }\n}\n\n// Subscribe يشترك بمعالج في حدث\nfunc (eb *EventBus) Subscribe(event string, handler func(string)) {\n    eb.mu.Lock()\n    defer eb.mu.Unlock()\n    eb.subscribers[event] = append(eb.subscribers[event], handler)\n}\n\n// Publish ينشر الحدث ويستدعي جميع المشتركين بشكل متزامن\nfunc (eb *EventBus) Publish(event string, data string) {\n    eb.mu.RLock()\n    defer eb.mu.RUnlock()\n    if handlers, ok := eb.subscribers[event]; ok {\n        for _, handler := range handlers {\n            handler(data)\n        }\n    }\n}\n\nfunc main() {\n    bus := NewEventBus()\n    \n    bus.Subscribe("user.login", func(data string) {\n        fmt.Printf("تسجيل الدخول: %s\\n", data)\n    })\n    \n    bus.Subscribe("user.login", func(data string) {\n        fmt.Printf("تدقيق: %s دخل إلى النظام\\n", data)\n    })\n    \n    bus.Publish("user.login", "alice@example.com")\n    // الإخراج:\n    // تسجيل الدخول: alice@example.com\n    // تدقيق: alice@example.com دخل إلى النظام\n}

هذا الكود يعمل، لكن هناك فارق بسيط: يتم استدعاء المعالجات بشكل متزامن. إذا توقف أحد المعالجات، فسيؤدي ذلك إلى حظر الناقل بأكمله.

الإصدار غير المتزامن مع القنوات

سنضيف عدم التزامن عبر القنوات المخزنة مؤقتًا. سيحصل كل مشترك على قناته الخاصة.

type AsyncEventBus struct {\n    subscribers map[string][]chan string\n    mu          sync.RWMutex\n}\n\nfunc NewAsyncEventBus() *AsyncEventBus {\n    return &AsyncEventBus{\n        subscribers: make(map[string][]chan string),\n    }\n}\n\nfunc (eb *AsyncEventBus) Subscribe(event string, buffer int) <-chan string {\n    eb.mu.Lock()\n    defer eb.mu.Unlock()\n    ch := make(chan string, buffer)\n    eb.subscribers[event] = append(eb.subscribers[event], ch)\n    return ch\n}\n\nfunc (eb *AsyncEventBus) Publish(event string, data string) {\n    eb.mu.RLock()\n    defer eb.mu.RUnlock()\n    if channe