Go (Golang) में Apache Kafka के साथ कार्य: लाइब्रेरीज़ और उदाहरण

Go (Golang) में Apache Kafka के साथ कार्य: लाइब्रेरीज़, उदाहरण और सर्वोत्तम अभ्यास

Apache Kafka एक वितरित डेटा स्ट्रीमिंग प्लेटफ़ॉर्म है, जो उच्च-लोड संदेश विनिमय प्रणालियों के निर्माण के लिए डी-फैक्टो मानक बन गया है। Go (Golang) इकोसिस्टम में Kafka के साथ इंटरैक्ट करने के लिए कई परिपक्व लाइब्रेरीज़ मौजूद हैं। इस लेख में हम दो सबसे लोकप्रिय लाइब्रेरीज़ पर चर्चा करेंगे: confluent-kafka-go (Confluent का आधिकारिक क्लाइंट) और sarama (शुद्ध Go क्लाइंट)। आप सीखेंगे कि प्रोड्यूसर और कंज्यूमर कैसे सेटअप करें, त्रुटियों को कैसे संभालें और अपने प्रोजेक्ट के लिए उपयुक्त टूल कैसे चुनें।

Kafka और Go एक बेहतरीन जोड़ी क्यों हैं?

Go को तेज़, समवर्ती और विश्वसनीय सर्वर एप्लिकेशन बनाने के लिए डिज़ाइन किया गया था। दूसरी ओर, Kafka को उच्च थ्रूपुट और न्यूनतम विलंबता की आवश्यकता होती है। Go गोरूटीन और चैनल संदेश स्ट्रीम के एसिंक्रोनस प्रोसेसिंग के लिए आदर्श हैं। इसके अलावा, Go की स्थिर टाइपिंग कंपाइलेशन चरण में कई त्रुटियों से बचने में मदद करती है, जो प्रोडक्शन सिस्टम के लिए महत्वपूर्ण है।

1. confluent-kafka-go लाइब्रेरी

यह C-क्लाइंट librdkafka के ऊपर एक रैपर है, जो अधिकतम प्रदर्शन और Kafka की सभी सुविधाओं (ट्रांज़ैक्शन, आइडेम्पोटेंसी, सुरक्षा प्रोटोकॉल) का पूर्ण समर्थन सुनिश्चित करता है। उच्च-लोड सिस्टम के लिए अनुशंसित।

स्थापना

सिस्टम में librdkafka की आवश्यकता है:

# macOSbrew install librdkafka

# Ubuntu/Debiansudo apt-get install librdkafka-dev

# फिर Go पैकेज स्थापित करेंgo 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] ऑफ़सेट %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) का उदाहरण

order-processors समूह में orders टॉपिक से संदेश पढ़ता है:

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] से", msg.Value, msg.Key, *msg.TopicPartition.Topic, msg.TopicPartition.Partition)