ساختن یک سیستم صف پیام قوی با کافکا و گوفام

go get github.com/IBM/sarama

# config.yaml
kafka:
  address: 
    - "localhost:9092"
  topic: "my_topic"

package main

import (
    "context"
    "github.com/gogf/gf/v2/frame/g"
    "github.com/gogf/gf/v2/os/gctx"
    "github.com/IBM/sarama"
)

var kafkaProducer sarama.SyncProducer

func initKafkaProducer(ctx context.Context) {
    // Create producer config
    config := sarama.NewConfig()
    config.Producer.Return.Successes = true

    // Get address from config
    address := g.Cfg().MustGet(ctx, "kafka.address").Strings()

    // Initialize producer
    producer, err := sarama.NewSyncProducer(address, config)
    if err != nil {
       panic(err)
    }
    kafkaProducer = producer
}

var kafkaConsumer sarama.Consumer

func initKafkaConsumer(ctx context.Context) {
    address := g.Cfg().MustGet(ctx, "kafka.address").Strings()

    consumer, err := sarama.NewConsumer(address, nil)
    if err != nil {
       panic(err)
    }
    kafkaConsumer = consumer
}

func sendMessage(ctx context.Context, message string) error {
    msg := &sarama.ProducerMessage{
        Topic: g.Cfg().MustGet(ctx, "kafka.topic").String(),
        Value: sarama.StringEncoder(message),
    }

    partition, offset, err := kafkaProducer.SendMessage(msg)
    if err != nil {
        return fmt.Errorf("failed to send message: %w", err)
    }

    g.Log().Infof(ctx, "Message sent successfully partition=%d, offset=%d", 
        partition, offset)
    return nil
}

func consumeMessages(ctx context.Context) {
    topic := g.Cfg().MustGet(ctx, "kafka.topic").String()
    partitionList, err := kafkaConsumer.Partitions(topic)
    if err != nil {
        g.Log().Errorf(ctx, "Failed to get partition list: %v", err)
        return
    }

    // Create a channel to handle shutdown
    done := make(chan bool)

    for partition := range partitionList {
        // Start a goroutine for each partition
        go func(partition int32) {
            pc, err := kafkaConsumer.ConsumePartition(topic, partition, 
                sarama.OffsetNewest)
            if err != nil {
                g.Log().Errorf(ctx, "Failed to start consumer: %v", err)
                return
            }

            defer pc.Close()

            for {
                select {
                case msg := <-pc.Messages():
                    handleMessageWithRetry(ctx, msg)
                case <-ctx.Done():
                    done <- true
                    return
                }
            }
        }(int32(partition))
    }

    <-done // Wait for shutdown signal
}

const (
    maxRetries = 3
    retryDelay = time.Second
)

func handleMessageWithRetry(ctx context.Context, msg *sarama.ConsumerMessage) {
    var err error
    for attempt := 0; attempt < maxRetries; attempt++ {
        err = processMessage(ctx, msg)
        if err == nil {
            // Success! Let's mark the message as processed
            markMessageProcessed(msg)
            return
        }

        g.Log().Warningf(ctx, 
            "Failed to process message (attempt %d/%d): %v", 
            attempt+1, maxRetries, err)

        if attempt < maxRetries-1 {
            time.Sleep(retryDelay * time.Duration(attempt+1))
        }
    }

    // If we get here, all retries failed
    g.Log().Errorf(ctx, 
        "Failed to process message after %d attempts: %v", 
        maxRetries, err)

    // Here you might want to:
    // 1. Send to a dead letter queue
    // 2. Store in an error log
    // 3. Trigger an alert
    handleFailedMessage(ctx, msg, err)
}

   # Start ZooKeeper first
   ./zookeeper-server-start.sh config/zookeeper.properties

   # Then start Kafka
   ./kafka-server-start.sh config/server.properties

   defer func() {
       if err := kafkaProducer.Close(); err != nil {
           g.Log().Errorf(ctx, "Failed to close producer: %v", err)
       }
       if err := kafkaConsumer.Close(); err != nil {
           g.Log().Errorf(ctx, "Failed to close consumer: %v", err)
       }
   }()