العمل مع Kafka في Go (Golang): مكتبة segmentio/kafka-go

العمل مع Kafka في Go (Golang) باستخدام مكتبة segmentio/kafka-go

Apache Kafka هي منصة موزعة لتدفق البيانات أصبحت المعيار الفعلي لبناء أنظمة تبادل الرسائل عالية التحميل. عند التطوير باستخدام Go (Golang)، تعد مكتبة segmentio/kafka-go واحدة من أشهر وأكثر المكتبات موثوقية للتفاعل مع Kafka. في هذه المقالة، سنشرح بالتفصيل كيفية تثبيت هذه المكتبة وإعدادها واستخدامها لإنشاء المنتجين والمستهلكين.

لماذا segmentio/kafka-go؟

توجد عدة مكتبات Go لـ Kafka في السوق (مثل Sarama و Confluent Go Client). تتميز segmentio/kafka-go بالمزايا التالية:

\n
• Go خالص — لا يتطلب تبعيات C (على عكس Confluent)، مما يبسط عملية البناء والنشر.
\n
• دعم بروتوكول Kafka v2 — ضغط فعال، تجميع (batching)، والعمل مع المعاملات.
\n
• API بسيط — واجهات بديهية للمنتجين والمستهلكين وإدارة المواضيع.
\n
• مجتمع نشط — تحديثات منتظمة وتوثيق جيد.
\n

تثبيت المكتبة

للبدء، أضف المكتبة إلى مشروعك عبر الأمر go get:

go get github.com/segmentio/kafka-go

تأكد من تشغيل مثيل Kafka (محليًا أو عبر Docker). مثال على تكوين Docker Compese الأدنى:

version: '3'\nservices:\n  zookeeper:\n    image: confluentinc/cp-zookeeper:latest\n    environment:\n      ZOOKEEPER_CLIENT_PORT: 2181\n  kafka:\n    image: confluentinc/cp-kafka:latest\n    depends_on:\n      - zookeeper\n    ports:\n      - "9092:9092"\n    environment:\n      KAFKA_BROKER_ID: 1\n      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181\n      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092

إنشاء منتج (Producer)

المنتج يرسل الرسائل إلى موضوع Kafka. لنلقِ نظرة على مثال أساسي:

package main\n\nimport (\n    "context"\n    "fmt"\n    "log"\n    "time"\n\n    "github.com/segmentio/kafka-go"\n)\n\nfunc main() {\n    // إعداد المنتج\n    writer := &kafka.Writer{\n        Addr:     kafka.TCP("localhost:9092"),\n        Topic:    "test-topic",\n        Balancer: &kafka.LeastBytes{}, // موازنة حسب حجم الرسالة\n        BatchSize: 10,                 // إرسال دفعة من 10 رسائل\n        BatchTimeout: 100 * time.Millisecond,\n    }\n    defer writer.Close()\n\n    // إرسال رسالة\n    err := writer.WriteMessages(context.Background(),\n        kafka.Message{\n            Key:   []byte("key-1"),\n            Value: []byte("Hello, Kafka from Go!"),\n        },\n    )\n    if err != nil {\n        log.Fatal("خطأ في إرسال الرسالة:", err)\n    }\n\n    fmt.Println("تم إرسال الرسالة بنجاح")\n}

المعلمات الرئيسية لـ kafka.Writer:

\n
• Addr — عنوان وسيط Kafka.
\n
• Topic — اسم الموضوع.
\n
• Balancer — استراتيجية توزيع الرسائل على الأقسام (LeastBytes, RoundRobin, Hash).
\n
• BatchSize و BatchTimeout — تحسين الأداء.
\n

إنشاء مستهلك (Consumer)

المستهلك يقرأ الرسائل من الموضوع. مثال بسيط مع إدارة يدوية للإزاحة (offset):

package main\n\nimport (\n    "context"\n    "fmt"\n    "log"\n\n    "github.com/segmentio/kafka-go"\n)\n\nfunc main() {\n    // إعداد القارئ (المستهلك)\n    reader := kafka.NewReader(kafka.ReaderConfig{\n        Brokers:   []string{"localhost:9092"},\n        Topic:     "test-topic",\n        GroupID:   "my-consumer-group",\n        MinBytes:  10, // الحد الأدنى لحجم الدفعة\n        MaxBytes:  10e6, // 10 ميجابايت\n        StartOffset: kafka.FirstOffset, // ابدأ من أول إزاحة\n