【kafka】简单运用go语言操作kafka实现生产者和消费者功能的包，confluent-kafka-go和sarama

kafkagosarama_0">confluent-kafka-go和sarama对比

特性	confluent-kafka-go	sarama
底层实现	基于 librdkafka C 库	完全用 Go 实现
性能	高吞吐量、低延迟	吞吐量较低，适合常规应用
安装依赖	需要 C 编译器和 librdkafka	无需外部依赖，纯 Go 实现
功能	支持 Kafka 所有功能，包括事务	支持 Kafka 核心功能，事务支持较弱
使用难度	配置复杂，需理解底层 C 库	使用简便，快速上手
社区支持	由 Confluent 官方支持	社区驱动，文档丰富
错误处理和日志	错误处理较为复杂，日志记录较为详细	错误处理简单，日志记录清晰
适用场景	高性能要求、高吞吐量的生产环境	一般的生产和消费场景，快速开发

基础使用案例

使用 confluent-kafka-go 发送消息
使用 confluent-kafka-go 库向 Kafka 发送消息。

package mainimport ("fmt""log""github.com/confluentinc/confluent-kafka-go/kafka"
)func main() {config := &kafka.ConfigMap{"bootstrap.servers": "localhost:9092", // Kafka服务器地址}producer, err := kafka.NewProducer(config)if err != nil {log.Fatal(err)}defer producer.Close()message := &kafka.Message{TopicPartition: kafka.TopicPartition{Topic: &"test_topic", Partition: kafka.PartitionAny},Value:          []byte("Hello Kafka from Go!"),}err = producer.Produce(message, nil)if err != nil {log.Fatal("Failed to produce message:", err)} else {fmt.Println("Message sent successfully!")}producer.Flush(15 * 1000)
}

使用 confluent-kafka-go 消费消息
使用 confluent-kafka-go 库从 Kafka 中消费消息。

package mainimport ("fmt""log""github.com/confluentinc/confluent-kafka-go/kafka"
)func main() {config := &kafka.ConfigMap{"bootstrap.servers": "localhost:9092", // Kafka服务器地址"group.id":          "test-group",     // 消费者组ID"auto.offset.reset": "earliest",        // 自动从最早的消息开始消费}consumer, err := kafka.NewConsumer(config)if err != nil {log.Fatal(err)}defer consumer.Close()err = consumer.Subscribe("test_topic", nil)if err != nil {log.Fatal("Failed to subscribe:", err)}for {msg, err := consumer.ReadMessage(-1)if err == nil {fmt.Printf("Consumed message: %s\n", string(msg.Value))} else {fmt.Printf("Error while consuming: %v\n", err)}}
}

使用 sarama 发送消息
使用 sarama 库向 Kafka 发送消息。

package mainimport ("fmt""log""github.com/Shopify/sarama"
)func main() {config := sarama.NewConfig()config.Producer.Return.Successes = trueproducer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)if err != nil {log.Fatal("Failed to create producer:", err)}defer producer.Close()message := &sarama.ProducerMessage{Topic: "test_topic",Value: sarama.StringEncoder("Hello Kafka from Go (Sarama)!"),}partition, offset, err := producer.SendMessage(message)if err != nil {log.Fatal("Failed to send message:", err)}fmt.Printf("Message sent to partition %d with offset %d\n", partition, offset)
}

使用 sarama 消费消息
使用 sarama 库从 Kafka 中消费消息。

package mainimport ("fmt""log""github.com/Shopify/sarama"
)func main() {config := sarama.NewConfig()config.Consumer.Return.Errors = trueconsumer, err := sarama.NewConsumer([]string{"localhost:9092"}, config)if err != nil {log.Fatal("Failed to create consumer:", err)}defer consumer.Close()partitions, err := consumer.Partitions("test_topic")if err != nil {log.Fatal("Failed to get partitions:", err)}for _, partition := range partitions {pc, err := consumer.ConsumePartition("test_topic", partition, sarama.OffsetNewest)if err != nil {log.Fatal("Failed to start consumer for partition:", err)}defer pc.Close()for msg := range pc.Messages() {fmt.Printf("Consumed message: %s\n", string(msg.Value))}}
}

性能案例对比

性能
confluent-kafka-go：
由于底层使用了 librdkafka，confluent-kafka-go 通常在吞吐量、延迟和连接管理方面表现得更加优越。
适合用于高吞吐量、低延迟的生产环境。

// 高吞吐量性能优化示例：
package mainimport ("fmt""github.com/confluentinc/confluent-kafka-go/kafka"
)func main() {config := &kafka.ConfigMap{"bootstrap.servers": "localhost:9092",}producer, _ := kafka.NewProducer(config)defer producer.Close()for i := 0; i < 10000; i++ {producer.Produce(&kafka.Message{TopicPartition: kafka.TopicPartition{Topic: &"test_topic", Partition: kafka.PartitionAny},Value:          []byte(fmt.Sprintf("Message %d", i)),}, nil)}producer.Flush(15 * 1000)
}

sarama：
虽然 sarama 的性能不及 confluent-kafka-go，但它对于大多数常规用途仍然足够快，特别是在吞吐量要求不是特别高的场景中。

// sarama 吞吐量示例：
package mainimport ("fmt""github.com/Shopify/sarama"
)func main() {config := sarama.NewConfig()config.Producer.Return.Successes = trueproducer, _ := sarama.NewSyncProducer([]string{"localhost:9092"}, config)defer producer.Close()for i := 0; i < 10000; i++ {producer.SendMessage(&sarama.ProducerMessage{Topic: "test_topic",Value: sarama.StringEncoder(fmt.Sprintf("Message %d", i)),})}
}

功能
confluent-kafka-go：
提供了丰富的功能，支持 Kafka 的所有核心功能，如生产者、消费者、消费者组管理、消息传递、事务支持、数据压缩等。
支持 Kafka 的最新特性，如消息事务、压缩、性能调优等。
由于是 librdkafka 的封装，confluent-kafka-go 与 Kafka 的版本兼容性更好，能够快速支持 Kafka 的新功能。

// 使用事务的生产者示例：
package mainimport ("fmt""log""github.com/confluentinc/confluent-kafka-go/kafka"
)func main() {config := &kafka.ConfigMap{"bootstrap.servers": "localhost:9092","acks":               "all",}producer, _ := kafka.NewProducer(config)defer producer.Close()// 开启事务producer.BeginTransaction()producer.Produce(&kafka.Message{TopicPartition: kafka.TopicPartition{Topic: &"test_topic", Partition: kafka.PartitionAny},Value:          []byte("Transactional Message"),}, nil)// 提交事务producer.CommitTransaction()
}

sarama：
提供了 Kafka 的核心功能，但可能在一些高级特性上不如 confluent-kafka-go 丰富。例如，sarama 对事务支持相对较弱，尽管在常规的生产/消费场景中功能足够。
支持 Kafka 的基本功能，如生产者、消费者组、消息传递等，但对一些高级功能（如流控、集群管理等）的支持可能稍有不足。

// sarama 事务支持相对较弱，但基本生产和消费功能已足够：
package mainimport ("fmt""log""github.com/Shopify/sarama"
)func main() {config := sarama.NewConfig()config.Producer.Return.Successes = trueproducer, _ := sarama.NewSyncProducer([]string{"localhost:9092"}, config)defer producer.Close()// 模拟事务：sarama 本身不直接支持事务，通常通过事务标记和重试来实现message := &sarama.ProducerMessage{Topic: "test_topic",Value: sarama.StringEncoder("Transactional Message"),}partition, offset, _ := producer.SendMessage(message)fmt.Printf("Message sent to partition %d with offset %d\n", partition, offset)
}