生产者与消费者

Go Producer

import "github.com/IBM/sarama"

func main() {
    config := sarama.NewConfig()
    config.Producer.RequiredAcks = sarama.WaitForAll
    config.Producer.Retry.Max = 5
    
    producer, _ := sarama.NewSyncProducer([]string{"localhost:9092"}, config)
    defer producer.Close()
    
    msg := &sarama.ProducerMessage{
        Topic: "my-topic",
        Value: sarama.StringEncoder("Hello Kafka"),
    }
    
    partition, offset, _ := producer.SendMessage(msg)
    fmt.Printf("Partition: %d, Offset: %d\n", partition, offset)
}

Go Consumer

func main() {
    consumer, _ := sarama.NewConsumer([]string{"localhost:9092"}, nil)
    defer consumer.Close()
    
    partitionConsumer, _ := consumer.ConsumePartition("my-topic", 0, sarama.OffsetNewest)
    defer partitionConsumer.Close()
    
    for msg := range partitionConsumer.Messages() {
        fmt.Printf("Message: %s\n", string(msg.Value))
    }
}

Consumer Group

import "github.com/IBM/sarama"

type ConsumerHandler struct{}

func (h *ConsumerHandler) Setup(sarama.ConsumerGroupSession) error   { return nil }
func (h *ConsumerHandler) Cleanup(sarama.ConsumerGroupSession) error { return nil }
func (h *ConsumerHandler) ConsumeClaim(session sarama.ConsumerGroupSession, 
    claim sarama.ConsumerGroupClaim) error {
    
    for msg := range claim.Messages() {
        fmt.Printf("Message: %s\n", string(msg.Value))
        session.MarkMessage(msg, "")
    }
    return nil
}

func main() {
    config := sarama.NewConfig()
    config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategyRoundRobin
    
    group, _ := sarama.NewConsumerGroup([]string{"localhost:9092"}, "my-group", config)
    handler := &ConsumerHandler{}
    
    for {
        group.Consume(context.Background(), []string{"my-topic"}, handler)
    }
}

参考：https://github.com/IBM/sarama