在 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] offset %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] of