العمل مع Apache Kafka في Go (Golang): المكتبات والأمثلة

العمل مع Apache Kafka في Go (Golang): المكتبات والأمثلة وأفضل الممارسات

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

لماذا يعتبر Kafka وGo ثنائيًا ممتازًا؟

تم إنشاء Go لبناء تطبيقات خادم سريعة ومتزامنة وموثوقة. من ناحية أخرى، يتطلب Kafka إنتاجية عالية وأقل تأخير. تعتبر Goroutines والقنوات في Go مثالية للمعالجة غير المتزامنة لتدفقات الرسائل. بالإضافة إلى ذلك، يساعد النوع الثابت في Go على تجنب العديد من الأخطاء في مرحلة الترجمة، وهو أمر بالغ الأهمية لأنظمة الإنتاج.

1. مكتبة confluent-kafka-go

هذه غلاف فوق عميل C librdkafka، مما يوفر أقصى أداء ودعمًا كاملاً لجميع ميزات Kafka (المعاملات، التماثل، بروتوكولات الأمان). يُوصى بها للأنظمة عالية التحميل.

التثبيت

يتطلب وجود librdkafka في النظام:

# macOSbrew install librdkafka

# Ubuntu/Debiansudo apt-get install librdkafka-dev

# ثم تثبيت حزمة Gogo get -u github.com/confluentinc/confluent-kafka-go/kafka

مثال على المنتج (Producer)

يرسل رسالة إلى الموضوع orders:

package main

import (    "fmt"    "github.com/confluentinc/confluent-kafka-go/kafka")

func main() {    p, err := kafka.NewProducer(&kafka.ConfigMap{        "bootstrap.servers": "localhost:9092",        "acks":              "all",    })    if err != nil {        panic(err)    }    defer p.Close()

    // قناة للإشعارات بالتسليم    go func() {        for e := range p.Events() {            switch ev := e.(type) {            case *kafka.Message:                if ev.TopicPartition.Error != nil {                    fmt.Printf("خطأ في التسليم: %v", ev.TopicPartition.Error)                } else {                    fmt.Printf("تم تسليم الرسالة إلى %s [%d] offset %d",                        *ev.TopicPartition.Topic, ev.TopicPartition.Partition, ev.TopicPartition.Offset)                }            }        }    }()

    topic := "orders"    message := []byte("طلب جديد #12345")    err = p.Produce(&kafka.Message{        TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},        Value:          message,        Key:            []byte("order_key"),    }, nil)

    if err != nil {        fmt.Printf("خطأ في الإرسال: %v", err)    }

    // انتظار إرسال جميع الرسائل    p.Flush(15 * 1000)}

مثال على المستهلك (Consumer)

يقرأ الرسائل من الموضوع orders في المجموعة order-processors:

package main

import (    "fmt"    "github.com/confluentinc/confluent-kafka-go/kafka")

func main() {    c, err := kafka.NewConsumer(&kafka.ConfigMap{        "bootstrap.servers": "localhost:9092",        "group.id":          "order-processors",        "auto.offset.reset": "earliest",    })    if err != nil {        panic(err)    }    defer c.Close()

    err = c.SubscribeTopics([]string{"orders"}, nil)    if err != nil {        panic(err)    }

    for {        msg, err := c.ReadMessage(-1) // انتظار غير محدود        if err != nil {            fmt.Printf("خطأ في الاستلام: %v", err)            continue        }        fmt.Printf("تم استلام الرسالة: %s (المفتاح: %s) من %s [%d] of