confluentinc/confluent-kafka-go

# Confluent 的 Apache Kafka® Golang 客户端 **confluent-kafka-go** 是 Confluent 提供的 [Apache Kafka](http://kafka.apache.org/) 和 [Confluent Platform](https://www.confluent.io/product/compare/) 的 Golang 客户端。 特性： - **高性能** - confluent-kafka-go 是一个轻量级封装，基于 [librdkafka](https://github.com/confluentinc/librdkafka)，这是一个经过精细优化的 C 客户端。 - **可靠性** - 编写 Apache Kafka 客户端需要处理大量的细节。我们在同一个地方（librdkafka）将其完美解决，并在我们所有的客户端（包括 [confluent-kafka-python](https://github.com/confluentinc/confluent-kafka-python) 和 [confluent-kafka-dotnet](https://github.com/confluentinc/confluent-kafka-dotnet)）中复用这些成果。 - **支持** - 由 [Confluent](https://confluent.io/) 提供商业支持。 - **面向未来** - 由 Kafka 的原作者/共同创建者创立的 Confluent 正在构建一个以 Apache Kafka 为核心的[流处理平台](https://www.confluent.io/product/compare/)。 对我们而言，保持客户端功能与核心 Apache Kafka 以及 [Confluent Platform](https://www.confluent.io/product/compare/) 组件同步是重中之重。 请参阅 [API 文档](http://docs.confluent.io/current/clients/confluent-kafka-go/index.html) 获取更多信息。 有关使用该客户端的分步指南，请参阅 [Apache Kafka 和 Golang 入门](https://developer.confluent.io/get-started/go/)。 # 示例 高级均衡消费者 ``` import ( "fmt" "time" "github.com/confluentinc/confluent-kafka-go/v2/kafka" ) func main() { c, err := kafka.NewConsumer(&kafka.ConfigMap{ "bootstrap.servers": "localhost", "group.id": "myGroup", "auto.offset.reset": "earliest", }) if err != nil { panic(err) } err = c.SubscribeTopics([]string{"myTopic", "^aRegex.*[Tt]opic"}, nil) if err != nil { panic(err) } // A signal handler or similar could be used to set this to false to break the loop. run := true for run { msg, err := c.ReadMessage(time.Second) if err == nil { fmt.Printf("Message on %s: %s\n", msg.TopicPartition, string(msg.Value)) } else if !err.(kafka.Error).IsTimeout() { // The client will automatically try to recover from all errors. // Timeout is not considered an error because it is raised by // ReadMessage in absence of messages. fmt.Printf("Consumer error: %v (%v)\n", err, msg) } } c.Close() } ``` 生产者 ``` import ( "fmt" "github.com/confluentinc/confluent-kafka-go/v2/kafka" ) func main() { p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": "localhost"}) if err != nil { panic(err) } defer p.Close() // Delivery report handler for produced messages go func() { for e := range p.Events() { switch ev := e.(type) { case *kafka.Message: if ev.TopicPartition.Error != nil { fmt.Printf("Delivery failed: %v\n", ev.TopicPartition) } else { fmt.Printf("Delivered message to %v\n", ev.TopicPartition) } } } }() // Produce messages to topic (asynchronously) topic := "myTopic" for _, word := range []string{"Welcome", "to", "the", "Confluent", "Kafka", "Golang", "client"} { p.Produce(&kafka.Message{ TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny}, Value: []byte(word), }, nil) } // Wait for message deliveries before shutting down p.Flush(15 * 1000) } ``` 在 [examples](examples) 目录中提供了更详尽的示例， 包括如何将 Go 客户端[配置为](examples/confluent_cloud_example)连接 [Confluent Cloud](https://www.confluent.io/confluent-cloud/)。 # 快速入门 支持 Go 1.17+ 和 librdkafka 2.15.0+。 ## 使用 Go Modules 你可以使用 [Go Modules](https://blog.golang.org/using-go-modules) 来安装 confluent-kafka-go。 在你的代码中从 GitHub 导入 `kafka` 包： ``` import "github.com/confluentinc/confluent-kafka-go/v2/kafka" ``` 构建你的项目： ``` go build ./... ``` 如果你正在为 Alpine Linux (musl) 构建，则必须指定 `-tags musl`。 ``` go build -tags musl ./... ``` 最新稳定版本的 confluent-kafka-go 依赖应该会自动添加到 你的 `go.mod` 文件中。 ## 安装客户端 手动安装： ``` go get -u github.com/confluentinc/confluent-kafka-go/v2/kafka ``` Golang 导入： ``` import "github.com/confluentinc/confluent-kafka-go/v2/kafka" ``` ## librdkafka 预编译的 librdkafka 二进制文件已包含在 Go 客户端中，无需在构建或目标系统上单独安装 librdkafka。 预编译的 librdkafka 二进制文件支持以下平台： * Mac OSX x64 和 arm64 * 基于 glibc 的 Linux x64 和 arm64（例如 RedHat、Debian、CentOS、Ubuntu 等） - 不支持 GSSAPI/Kerberos * 基于 musl 的 Linux amd64 和 arm64 (Alpine) - 不支持 GSSAPI/Kerberos * Windows amd64 - 不支持 GSSAPI/Kerberos 在为 Alpine Linux (musl libc) 构建你的应用程序时，你必须向 `go get`、`go build` 等传递 `-tags musl` 参数。 `CGO_ENABLED` 不能设置为 `0`，因为 Go 客户端是基于 C 库 librdkafka 的。 ## 安装 librdkafka 如果内置的 librdkafka 构建不支持你的平台，或者 你需要支持 GSSAPI/Kerberos 的 librdkafka，你必须使用以下替代方案之一在构建和目标系统上手动安装 librdkafka： - 对于基于 Debian 和 Ubuntu 的发行版，从标准仓库安装 `librdkafka-dev` 或使用 [Confluent 的 Deb 仓库](http://docs.confluent.io/current/installation.html#installation-apt)。 - 对于基于 Redhat 的发行版，使用 [Confluent 的 YUM 仓库](http://docs.confluent.io/current/installation.html#rpm-packages-via-yum) 安装 `librdkafka-devel`。 - 对于 MacOS X，从 Homebrew 安装 `librdkafka`。如果你还没有安装它，你可能还需要通过 brew 安装 pkg-config：`brew install librdkafka pkg-config`。 - 对于 Alpine：`apk add librdkafka-dev pkgconf` - 对于 Windows：没有官方/受支持的软件包，但包含了用于 Windows/x64 的静态构建版本。 仅当需要支持 GSSAPI/Kerberos 时才需要从源码安装。 - 有关源码构建的说明，请参阅下文。 从源码构建： ``` git clone https://github.com/confluentinc/librdkafka.git cd librdkafka ./configure make sudo make install ``` 安装 librdkafka 后，你需要使用 `-tags dynamic` 构建你的 Go 应用程序。 **注意：** 如果你使用 Go 客户端的 `master` 分支，那么你需要使用 librdkafka 的 `master` 分支。 **confluent-kafka-go 需要 librdkafka v1.9.0 或更高版本。** ## Linux 上的静态构建 由于我们使用的是 `cgo`，即使使用了预编译、静态编译的 librdkafka（如 **librdkafka** 章节所述），Go 构建出的依然是一个动态链接库。 对于基于 `glibc` 的系统，如果编译客户端的系统与目标系统不同，尤其是当目标系统较旧时，在尝试运行编译后的客户端时会报错 `glibc` 版本错误。 遗憾的是，如果我们尝试构建静态链接的二进制文件，这并不能解决问题，因为使用 `glibc` 无法实现真正的静态构建。这是 因为在 `glibc` 中有一些函数（例如 `getaddrinfo`），即使代码是静态编译的，它们也需要该库的共享版本。 解决这个问题的一种方法是使用容器/虚拟机来构建二进制文件，或者在编译客户端的系统上安装 较旧版本的 `glibc`。 另一种方法是使用 `musl` 为 Linux 创建真正的静态构建。为此， [为你的系统安装它](https://wiki.musl-libc.org/getting-started.html)。 静态编译命令，旨在与预编译的 librdkafka 包一起使用： ``` CC=/path/to/musl-gcc go build --ldflags '-linkmode external -extldflags "-static"' -tags musl ``` # FIPS 140-3 合规性 当使用 Go 1.24.3 或更新版本时，此客户端支持 Schema Registry 操作的 FIPS 140-3 合规性。 ## 客户端与 Schema Registry 之间的通信 ``` GOFIPS140=inprocess go build -o myapp ``` **在 FIPS 模式下运行：** ``` GODEBUG=fips140=only ./myapp ``` 当使用 `GODEBUG=fips140=only` 运行时，应用程序将仅对到 Schema Registry 的所有 TLS 连接使用经过 FIPS 140-3 验证的加密实现。如果尝试任何非 FIPS 批准的加密操作，应用程序将立即发生 panic。 # API 体系 推荐的 API 体系是基于函数的（Function-Based）体系， 基于通道的（Channel-Based）体系在 [examples/legacy](examples/legacy) 中进行了说明。 ## 基于函数的消费者 消息、错误和事件通过 `consumer.Poll()` 函数进行轮询。 它与底层的 librdkafka 功能直接映射。 请参阅 [examples/consumer_example](examples/consumer_example) ## 基于函数的生产者 应用程序调用 `producer.Produce()` 来生产消息。 投递报告通过 `producer.Events()` 或指定的私有通道发出。 _警告_ * `Produce()` 是一个非阻塞调用，如果内部的 librdkafka 队列已满， 该调用将失败，可以进行重试。 请参阅 [examples/producer_example](examples/producer_example) # 许可证 [Apache License v2.0](http://www.apache.org/licenses/LICENSE-2.0) KAFKA 是 The Apache Software Foundation 的注册商标，并已授权给 confluent-kafka-go 使用。confluent-kafka-go 与 The Apache Software Foundation 没有任何隶属关系，也未获得其认可。 # 开发者须知 请参阅 [kafka/README](kafka/README.md) 非常欢迎对代码、示例、文档等方面的贡献。 进行你的更改，运行 `gofmt`、测试等，推送你的分支，创建 PR，并[签署 CLA](http://clabot.confluent.io/cla)。 # Confluent Cloud 有关将 Golang 客户端与 Confluent Cloud 结合使用的分步指南，请参阅 [Confluent Developer](https://developer.confluent.io/) 上的 [Apache Kafka 和 Golang 入门](https://developer.confluent.io/get-started/go/)。

标签：Golang, Kafka, SonarQube插件, 中间件, 后端开发, 安全编程, 客户端, 数据流处理, 日志审计, 消息队列, 软件成分分析