Golang 语言中 kafka 客户端库 sarama

简介: Golang 语言中 kafka 客户端库 sarama

介绍

Apache Kafka 是一款开源的消息引擎系统。它在项目中的作用主要是削峰填谷和解耦。本文我们只介绍 Apache Kafka 的 Golang 客户端库 Sarama。Sarama 是 MIT 许可的 Apache Kafka 0.8 及更高版本的 Golang 客户端库。

如果读者朋友对 Apache Kafka 服务端还不了解,建议先阅读官方文档中的入门部分,本文使用的版本是 Apache Kafka 2.8。

生产者

我们可以使用 Sarama 库的 AsyncProducer 或 SyncProducer 生产消息。在大多数情况下首选使用 AsyncProducer 生产消息。它通过一个 channel 接收消息,并在后台尽可能高效的异步生产消息。

SyncProducer 发送 Kafka 消息后阻塞,直到接收到 ACK 确认。SyncProducer 有两个警告:它通常效率较低,并且实际的耐用性保证取决于 Producer.RequiredAcks 的配置值。在某些配置中,有时仍会丢失由 SyncProducer 确认的消息,但是使用比较简单。

为了读者朋友们容易理解,本文我们介绍 SyncProducer 作为生产者的使用方式。如果读者朋友想了解 AsyncProducer 作为生产者的使用方式,请参考官方文档。

使用 SyncProducer 作为生产者的示例代码:

func sendMessage (brokerAddr []string, config *sarama.Config, topic string, value sarama.Encoder) {
 producer, err := sarama.NewSyncProducer(brokerAddr, config)
 if err != nil {
  fmt.Println(err)
  return
 }
 defer func() {
  if err = producer.Close(); err != nil {
   fmt.Println(err)
   return
  }
 }()
 msg := &sarama.ProducerMessage{
  Topic: topic,
  Value: value,
 }
 partition, offset, err := producer.SendMessage(msg)
 if err != nil {
  fmt.Println(err)
  return
 }
 fmt.Printf("partition:%d offset:%d\n", partition, offset)
}

阅读上面这段代码,我们调用 NewSyncProducer() 创建一个新的 SyncProducer,给定 broker 地址和配置信息。调用 SendMessage() 生产给定的消息,并且仅在生产成功或失败时返回。它将返回分区(Partition)和生产的消息的偏移量(Offset),如果消息生产失败,则返回错误。

需要注意的是,为了避免泄露,必须在生产者上调用 Close(),因为当它超出范围时,可能不会自动垃圾回收。

03

消费者

我们可以使用 Sarama 库的消费者 Consumer 或消费者组 ConsumerGroup API 消费消息。为了读者朋友们容易理解,本文我们介绍使用 Consumer 消费消息。

Consumer 管理 PartitionConsumers,该 PartitionConsumers 处理来自 brokers 的 Kafka 消息。

Consumer 消费消息的示例代码:

func consumer (brokenAddr []string, topic string, partition int32, offset int64) {
 consumer, err := sarama.NewConsumer(brokenAddr, nil)
 if err != nil {
  fmt.Println(err)
  return
 }
 defer func() {
  if err = consumer.Close(); err != nil {
   fmt.Println(err)
   return
  }
 }()
 partitionConsumer, err := consumer.ConsumePartition(topic, partition, offset)
 if err != nil {
  fmt.Println(err)
  return
 }
 defer func() {
  if err = partitionConsumer.Close(); err != nil {
   fmt.Println(err)
   return
  }
 }()
 for msg := range partitionConsumer.Messages() {
  fmt.Printf("partition:%d offset:%d key:%s val:%s\n", msg.Partition, msg.Offset, msg.Key, msg.Value)
 }
}

阅读上面这段代码,我们调用 NewConsumer() 创建一个新的 consumer,给定 broker 地址和配置信息。调用 ConsumePartition() 创建 PartitionConsumer,给定 topic、partition 和 offset。PartitionConsumer 处理来自给定 topic 和 partition 的 Kafka 消息。

需要注意的是,为了防止泄露,必须调用 consumer 和 partitionConsumer 的 Close(),因为当它超出范围时,可能不会自动垃圾回收。

04

总结

本文主要介绍如何使用 Apache Kafka 的 Golang 语言客户端库 Sarama 生产和消费 Kafka 消息。关于生产者和消费者,分别列举了一个简单示例。除此之外,Sarama 库还提供了很多其它 Api,感兴趣的读者朋友可以阅读官方文档了解更多。

推荐阅读:

参考资料:

https://github.com/Shopify/saramahttps://shopify.github.io/sarama/https://pkg.go.dev/github.com/Shopify/saramahttps://github.com/Shopify/sarama/wiki/Frequently-Asked-Questionshttps://kafka.apache.org/documentation/


目录
相关文章
|
6月前
|
Go
Golang语言之管道channel快速入门篇
这篇文章是关于Go语言中管道(channel)的快速入门教程,涵盖了管道的基本使用、有缓冲和无缓冲管道的区别、管道的关闭、遍历、协程和管道的协同工作、单向通道的使用以及select多路复用的详细案例和解释。
193 4
Golang语言之管道channel快速入门篇
|
6月前
|
Go
Golang语言之gRPC程序设计示例
这篇文章是关于Golang语言使用gRPC进行程序设计的详细教程,涵盖了RPC协议的介绍、gRPC环境的搭建、Protocol Buffers的使用、gRPC服务的编写和通信示例。
161 3
Golang语言之gRPC程序设计示例
|
20天前
|
网络协议 测试技术 Linux
Golang 实现轻量、快速的基于 Reactor 模式的非阻塞 TCP 网络库
gev 是一个基于 epoll 和 kqueue 实现的高性能事件循环库,适用于 Linux 和 macOS(Windows 暂不支持)。它支持多核多线程、动态扩容的 Ring Buffer 读写缓冲区、异步读写和 SO_REUSEPORT 端口重用。gev 使用少量 goroutine,监听连接并处理读写事件。性能测试显示其在不同配置下表现优异。安装命令:`go get -u github.com/Allenxuxu/gev`。
|
6月前
|
安全 Go
Golang语言goroutine协程并发安全及锁机制
这篇文章是关于Go语言中多协程操作同一数据问题、互斥锁Mutex和读写互斥锁RWMutex的详细介绍及使用案例,涵盖了如何使用这些同步原语来解决并发访问共享资源时的数据安全问题。
122 4
|
4月前
|
JSON Go 开发者
go-carbon v2.5.0 发布,轻量级、语义化、对开发者友好的 golang 时间处理库
carbon 是一个轻量级、语义化、对开发者友好的 Golang 时间处理库,提供了对时间穿越、时间差值、时间极值、时间判断、星座、星座、农历、儒略日 / 简化儒略日、波斯历 / 伊朗历的支持。
94 4
|
4月前
|
存储 Cloud Native Shell
go库介绍:Golang中的Viper库
Viper 是 Golang 中的一个强大配置管理库,支持环境变量、命令行参数、远程配置等多种配置来源。本文详细介绍了 Viper 的核心特点、应用场景及使用方法,并通过示例展示了其强大功能。无论是简单的 CLI 工具还是复杂的分布式系统,Viper 都能提供优雅的配置管理方案。
120 6
|
6月前
|
Prometheus Cloud Native Go
Golang语言之Prometheus的日志模块使用案例
这篇文章是关于如何在Golang语言项目中使用Prometheus的日志模块的案例,包括源代码编写、编译和测试步骤。
109 3
Golang语言之Prometheus的日志模块使用案例
|
5月前
|
前端开发 中间件 Go
实践Golang语言N层应用架构
【10月更文挑战第2天】本文介绍了如何在Go语言中使用Gin框架实现N层体系结构,借鉴了J2EE平台的多层分布式应用程序模型。文章首先概述了N层体系结构的基本概念,接着详细列出了Go语言中对应的构件名称,包括前端框架(如Vue.js、React)、Gin的处理函数和中间件、依赖注入和配置管理、会话管理和ORM库(如gorm或ent)。最后,提供了具体的代码示例,展示了如何实现HTTP请求处理、会话管理和数据库操作。
75 0
|
6月前
|
Go
Golang语言文件操作快速入门篇
这篇文章是关于Go语言文件操作快速入门的教程,涵盖了文件的读取、写入、复制操作以及使用标准库中的ioutil、bufio、os等包进行文件操作的详细案例。
94 4
Golang语言文件操作快速入门篇
|
6月前
|
Go
Golang语言错误处理机制
这篇文章是关于Golang语言错误处理机制的教程,介绍了使用defer结合recover捕获错误、基于errors.New自定义错误以及使用panic抛出自定义错误的方法。
75 3