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 main
import (
"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 main
import (
"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 main
import (
"fmt"
"log"
"github.com/Shopify/sarama"
)
func main() {
config := sarama.NewConfig()
config.Producer.Return.Successes = true
producer, 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 main
import (
"fmt"
"log"
"github.com/Shopify/sarama"
)
func main() {
config := sarama.NewConfig()
config.Consumer.Return.Errors = true
consumer, 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 main
import (
"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 main
import (
"fmt"
"github.com/Shopify/sarama"
)
func main() {
config := sarama.NewConfig()
config.Producer.Return.Successes = true
producer, _ := 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 main
import (
"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 main
import (
"fmt"
"log"
"github.com/Shopify/sarama"
)
func main() {
config := sarama.NewConfig()
config.Producer.Return.Successes = true
producer, _ := 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)
}
- 总结：
如果你需要高性能和 Kafka 高级特性，选择 confluent-kafka-go。
如果你追求易用性和快速开发，或者不希望依赖 C 库，选择 sarama。
-
选择 confluent-kafka-go：适用于高性能、高吞吐量的场景。
-
选择 sarama：适用于不需要复杂配置和高级特性的场景。
