Sarama Go: دليل شامل مع أمثلة عملية

ما هي Sarama ولماذا تحتاجها؟

Sarama هي مكتبة Go عالية الأداء ونقية للعمل مع Apache Kafka. توفر مجموعة كاملة من الأدوات للتفاعل مع مجموعة Kafka، بما في ذلك المنتجين (producers) لإرسال الرسائل، والمستهلكين (consumers) لقراءة الرسائل، بالإضافة إلى وظائف إدارية لإدارة المواضيع (topics) والأقسام (partitions) والإعدادات.

Apache Kafka هي منصة موزعة لتدفق البيانات تُستخدم لبناء خطوط أنابيب بيانات في الوقت الفعلي، وأنظمة تبادل الرسائل، ومعالجة الأحداث، وتخزين السجلات. تسمح Sarama لتطبيقات Go بالتكامل بسهولة مع Kafka، مما يضمن نقل بيانات موثوقًا وفعالًا.

لماذا تستخدم Sarama؟

• Go نقي: Sarama مكتوبة بالكامل بلغة Go دون أي تبعيات خارجية على C/C++، مما يبسط عملية البناء والنشر.
• أداء عالٍ: المكتبة محسّنة للعمل مع كميات كبيرة من البيانات وتدعم الإرسال غير المتزامن للرسائل.
• مرونة: تدعم أوضاع تشغيل مختلفة للمنتجين (متزامن، غير متزامن) والمستهلكين (بسيط، مجموعة مستهلكين).
• مجتمع نشط: Sarama هي واحدة من أشهر مكتبات Kafka للغة Go مع عدد كبير من المساهمين ووثائق شاملة.
• تحكم كامل: تمنح المطور تحكمًا كاملاً في الإعدادات، بما في ذلك إدارة الإزاحات (offsets)، والتسلسل/إلغاء التسلسل، ومعالجة الأخطاء.

التثبيت

يتم تثبيت Sarama في مشروع Go بالطريقة القياسية عبر مدير الحزم go get.

go get github.com/IBM/sarama

بعد تنفيذ هذا الأمر، ستتم إضافة المكتبة إلى ملف go.mod الخاص بك، وستتمكن من استيرادها في الكود الخاص بك.

ملاحظة مهمة: تأكد من تثبيت Go الإصدار 1.18 أو أحدث، حيث تستخدم Sarama ميزات اللغة الحديثة.

بداية سريعة: إرسال واستقبال الرسائل

لنكتب مثالًا بسيطًا يرسل رسالة إلى موضوع (topic) في Kafka ويقرأها.

1. المنتج (Producer) — إرسال رسالة

package main

import (
    "fmt"
    "log"

    "github.com/IBM/sarama"
)

func main() {
    // 1. إعدادات المنتج
    config := sarama.NewConfig()
    config.Producer.Return.Successes = true // مطلوب للوضع المتزامن
    config.Producer.RequiredAcks = sarama.WaitForAll // انتظار تأكيد من جميع النسخ المتماثلة

    // 2. إنشاء منتج متزامن
    producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)
    if err != nil {
        log.Fatalf("فشل إنشاء المنتج: %v", err)
    }
    defer producer.Close()

    // 3. إنشاء رسالة
    msg := &sarama.ProducerMessage{
        Topic: "my-topic",
        Key:   sarama.StringEncoder("my-key"),
        Value: sarama.StringEncoder("مرحبًا، Kafka!"),
    }

    // 4. إرسال الرسالة
    partition, offset, err := producer.SendMessage(msg)
    if err != nil {
        log.Fatalf("فشل إرسال الرسالة: %v", err)
    }

    fmt.Printf("تم إرسال الرسالة إلى القسم %d مع الإزاحة %d\n", partition, offset)
}

2. المستهلك (Consumer) — استقبال رسالة

package main

import (
    "fmt"
    "log"
    "os"
    "os/signal"
    "syscall"

    "github.com/IBM/sarama"
)

func main() {
    // 1. إعدادات المستهلك
    config := sarama.NewConfig()
    config.Consumer.Return.Errors = true
    config.Consumer.Offsets.Initial = sarama.OffsetOldest // القراءة من البداية

    // 2. إنشاء مستهلك
    consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, config)
    if err != nil {
        log.Fatalf("فشل إنشاء المستهلك: %v", err)
    }
    defer consumer.Close()

    // 3. الاشتراك في قسم
    parti