go-kafka

go kafka包

本文使用的是kafka-go 6.5k 这个包 其他包参考：

我们在细分市场中非常依赖GO和Kafka。不幸的是，在撰写本文时，Kafka的GO客户库的状态并不理想。可用选项是：

萨拉玛（Sarama） 10k，这是迄今为止最受欢迎的，但很难与之合作。它的记录不足，API暴露了Kafka协议的低级概念，并且不支持诸如上下文之类的GO。它还将所有值传递给引起大量动态内存分配，更频繁的垃圾收集和更高的内存使用情况的指针。

Confluent-kafka-Go 4.4k是围绕librdkafka的基于CGO的包装器，这意味着它将使用该软件包的所有GO代码引入了C库的依赖关系。它的文档比Sarama要好得多，但仍然缺乏对GO上下文的支持。

Goka2.2k 是GO的最新Kafka客户端，专注于特定的用法模式。它提供了将KAFKA用作服务之间的消息传递的消息，而不是事件的有序日志，但这并不是我们在细分市场的典型用例。该包还取决于与Kafka的所有互动的Sarama。

这就是Kafka-Go发挥作用的地方。它提供了低水平和高级API，可与Kafka进行交互，反映概念并实现GO标准库的接口，以使其易于使用并与现有软件集成。

go操作kafka

 go get github.com/segmentio/kafka-go

简单的生产者：

package mainimport ("context""github.com/segmentio/kafka-go""log""time"
)func main() {// to produce messagestopic := "my-topic"partition := 0//默认没有主题会创建conn, err := kafka.DialLeader(context.Background(), "tcp", "192.168.59.131:9092", topic, partition)if err != nil {log.Fatal("failed to dial leader:", err)}conn.SetWriteDeadline(time.Now().Add(10 * time.Second))_, err = conn.WriteMessages(kafka.Message{Value: []byte("one!")},kafka.Message{Value: []byte("two!")},kafka.Message{Value: []byte("three!")},)if err != nil {log.Fatal("failed to write messages:", err)}if err := conn.Close(); err != nil {log.Fatal("failed to close writer:", err)}
}

简单的消费者：

package mainimport ("context""fmt""github.com/segmentio/kafka-go""log""time"
)func main() {// to consume messagestopic := "my-topic"partition := 0conn, err := kafka.DialLeader(context.Background(), "tcp", "192.168.59.131:9092", topic, partition)if err != nil {log.Fatal("failed to dial leader:", err)}conn.SetReadDeadline(time.Now().Add(10 * time.Second))batch := conn.ReadBatch(10e3, 1e6) // fetch 10KB min, 1MB maxb := make([]byte, 10e3) // 10KB max per messagefor {n, err := batch.Read(b)if err != nil {break}fmt.Println(n, string(b[:n]))}if err := batch.Close(); err != nil {log.Fatal("failed to close batch:", err)}if err := conn.Close(); err != nil {log.Fatal("failed to close connection:", err)}
}

输出

4 one!
4 two!
6 three!
2023/08/22 16:48:02 failed to close batch:[7] Request Timed Out: the request exceeded the user-specified time limit in the request
exit status 1

连接

会自动创建主题：

// to create topics when auto.create.topics.enable='true' 
conn, err := kafka.DialLeader(context.Background(), "tcp", "localhost:9092", "my-topic", 0)
if err != nil {panic(err.Error())
}

连接 手动创建主题：

package mainimport ("github.com/segmentio/kafka-go""net""strconv"
)// kafka
func main() {// to create topics when auto.create.topics.enable='false'topic := "my-topic"conn, err := kafka.Dial("tcp", "192.168.59.131:9092")if err != nil {panic(err.Error())}defer conn.Close()controller, err := conn.Controller()if err != nil {panic(err.Error())}var controllerConn *kafka.ConncontrollerConn, err = kafka.Dial("tcp", net.JoinHostPort(controller.Host, strconv.Itoa(controller.Port)))if err != nil {panic(err.Error())}defer controllerConn.Close()topicConfigs := []kafka.TopicConfig{{Topic:             topic,NumPartitions:     1,ReplicationFactor: 1,},}err = controllerConn.CreateTopics(topicConfigs...)if err != nil {panic(err.Error())}//---------------读取主题------------------pl, err := conn.ReadPartitions()if err != nil {panic(err.Error())}for _, p := range pl {fmt.Println(p.Topic)}
}

Reader 消费者

官网介绍：
A Reader is another concept exposed by the kafka-go package, which intends to make it simpler to implement the typical use case of consuming from a single topic-partition pair. A Reader also automatically handles reconnections and offset management, and exposes an API that supports asynchronous cancellations and timeouts using Go contexts.

Note that it is important to call Close() on a Reader when a process exits. The kafka server needs a graceful disconnect to stop it from continuing to attempt to send messages to the connected clients. The given example will not call Close() if the process is terminated with SIGINT (ctrl-c at the shell) or SIGTERM (as docker stop or a kubernetes restart does). This can result in a delay when a new reader on the same topic connects (e.g. new process started or new container running). Use a signal.Notify handler to close the reader on process shutdown.
阅读器（Reader）是 kafka-go 软件包暴露的另一个概念，它旨在简化从单个主题-分区对中消费的典型用例的实现。阅读器还会自动处理重新连接和偏移管理，并提供一个 API，使用 Go 上下文支持异步取消和超时。

需要注意的是，在进程退出时调用 Reader 上的 Close() 非常重要。kafka 服务器需要优雅地断开连接，以阻止它继续尝试向已连接的客户端发送消息。如果进程被 SIGINT（在 shell 中按 ctrl-c）或 SIGTERM（docker stop 或 kubernetes 重启）终止，给出的示例将不会调用 Close()。当同一主题上有新的阅读器连接时（例如，新进程启动或新容器运行），这可能会导致延迟。使用 signal.Notify 处理程序在进程关闭时关闭阅读器。

package mainimport ("context""fmt""github.com/segmentio/kafka-go""log"
)// kafka
func main() {// make a new reader that consumes from topic-A, partition 0, at offset 42r := kafka.NewReader(kafka.ReaderConfig{Brokers:   []string{"192.168.59.131:9092", "192.168.59.131:9093", "192.168.59.131:9094"},Topic:     "topic-A",Partition: 0,MaxBytes:  10e6, // 10MB})r.SetOffset(4)fmt.Println("start")for {m, err := r.ReadMessage(context.Background())if err != nil {break}fmt.Printf("message at offset %d: %s = %s\n", m.Offset, string(m.Key), string(m.Value))}if err := r.Close(); err != nil {log.Fatal("failed to close reader:", err)}
}

Consumer Groups 消费者组

kafka-go 还支持 Kafka 消费者组，包括代理管理的偏移量。要启用消费者组，只需在 ReaderConfig 中指定 GroupID。
使用消费者组时，ReadMessage 会自动提交偏移量

package mainimport ("context""fmt""github.com/segmentio/kafka-go""log"
)// kafka
func main() {// make a new reader that consumes from topic-A, partition 0, at offset 42r := kafka.NewReader(kafka.ReaderConfig{Brokers:   []string{"192.168.59.131:9092", "192.168.59.131:9093", "192.168.59.131:9094"},Topic:     "topic-A",Partition: 0,MaxBytes:  10e6, // 10MBGroupID:   "AAA",})fmt.Println("start")for {m, err := r.ReadMessage(context.Background())if err != nil {break}fmt.Printf("message at offset %d: %s = %s\n", m.Offset, string(m.Key), string(m.Value))}if err := r.Close(); err != nil {log.Fatal("failed to close reader:", err)}
}

使用消费者组时有一些限制：
(*Reader).SetOffset 在设置 GroupID 时会返回错误信息
(*Reader).Offset在设置GroupID时将始终返回-1
(*Reader).Lag在GroupID被设置时总是返回-1
(*Reader).ReadLag 将在设置 GroupID 时返回错误信息
(*Reader).Stats在GroupID被设置时将返回-1的分区。

Explicit Commits 显式提交

没 CommitMessages 提交偏移量没变下次还会读到

package mainimport ("context""fmt""github.com/segmentio/kafka-go""log"
)// kafka
func main() {r := kafka.NewReader(kafka.ReaderConfig{Brokers:   []string{"192.168.59.131:9092", "192.168.59.131:9093", "192.168.59.131:9094"},Topic:     "topic-A",Partition: 0,MaxBytes:  10e6, // 10MBGroupID:   "AAA",})fmt.Println("start")ctx := context.Background()for {m, err := r.FetchMessage(ctx)if err != nil {break}fmt.Printf("message at topic/partition/offset %v/%v/%v: %s = %s\n", m.Topic, m.Partition, m.Offset, string(m.Key), string(m.Value))if err := r.CommitMessages(ctx, m); err != nil {log.Fatal("failed to commit messages:", err)}}if err := r.Close(); err != nil {log.Fatal("failed to close reader:", err)}
}

Managing Commits 管理提交 周期性提交

通过在 ReaderConfig 上设置 CommitInterval 来周期性地向 Kafka 提交偏移量。

r := kafka.NewReader(kafka.ReaderConfig{Brokers: []string{"192.168.59.131:9092", "192.168.59.131:9093", "192.168.59.131:9094"},Topic:   "topic-A",Partition:      0,MaxBytes:       10e6, // 10MBGroupID:        "AAA",CommitInterval: time.Second,// flushes commits to Kafka every second})

Writer 生产者

To produce messages to Kafka, a program may use the low-level Conn API, but the package also provides a higher level Writer type which is more appropriate to use in most cases as it provides additional features:

  • Automatic retries and reconnections on errors.
  • Configurable distribution of messages across available partitions.
  • Synchronous or asynchronous writes of messages to Kafka.
  • Asynchronous cancellation using contexts.
  • Flushing of pending messages on close to support graceful shutdowns.
  • Creation of a missing topic before publishing a message. Note! it was the default behaviour up to the version v0.4.30

要向 Kafka 发送消息，程序可以使用底层的 Conn API，但软件包也提供了更高级的 Writer 类型，在大多数情况下更适合使用，因为它提供了更多的功能：

  • 出错时自动重试和重新连接。
  • 可配置的可用分区信息分配。
  • 将消息同步或异步写入 Kafka。
  • 使用上下文进行异步取消。
  • 在关闭时刷新待处理消息，以支持优雅关机。
  • 在发布消息前创建缺失的主题。注意！这是 v0.4.30 之前的默认行为。
package mainimport ("context""github.com/segmentio/kafka-go""log"
)func main() {// make a writer that produces to topic-A, using the least-bytes distribution// 使用最少字节分布，制作向主题-A 发送信息的写入器w := &kafka.Writer{Addr:     kafka.TCP("192.168.59.131:9092", "192.168.59.131:9093", "192.168.59.131:9094"),Topic:    "topic-A",Balancer: &kafka.LeastBytes{},}err := w.WriteMessages(context.Background(),kafka.Message{Key:   []byte("Key-A"),Value: []byte("Hello World!One!"),},kafka.Message{Key:   []byte("Key-B"),Value: []byte("Hello World!Two!"),},kafka.Message{Key:   []byte("Key-C"),Value: []byte("Hello World!Three!"),},)if err != nil {log.Fatal("failed to write messages:", err)}if err := w.Close(); err != nil {log.Fatal("failed to close writer:", err)}
}

自动创建主题

package mainimport ("context""errors""github.com/segmentio/kafka-go""log""time"
)func main() {// Make a writer that publishes messages to topic-A.// The topic will be created if it is missing.w := &kafka.Writer{Addr:                   kafka.TCP("192.168.59.131:9092", "192.168.59.131:9093", "192.168.59.131:9094"),Topic:                  "topic-A",AllowAutoTopicCreation: true,}messages := []kafka.Message{{Key:   []byte("Key-A"),Value: []byte("Hello One!"),},{Key:   []byte("Key-B"),Value: []byte("Hello Two!"),},{Key:   []byte("Key-C"),Value: []byte("Hello Three!"),},}var err errorconst retries = 3for i := 0; i < retries; i++ {ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)defer cancel()// 尝试在发布消息前创建主题err = w.WriteMessages(ctx, messages...)if errors.Is(err, kafka.LeaderNotAvailable) || errors.Is(err, context.DeadlineExceeded) {time.Sleep(time.Millisecond * 250)continue}if err != nil {log.Fatalf("unexpected error %v", err)}break}if err := w.Close(); err != nil {log.Fatal("failed to close writer:", err)}
}

多个主题

通常，WriterConfig.Topic 用于初始化单主题写入器。通过排除该特定配置，您可以通过设置 Message.Topic.WriterConfig.Topic 来按消息定义主题。

package mainimport ("context""github.com/segmentio/kafka-go""log"
)func main() {w := &kafka.Writer{Addr: kafka.TCP("192.168.59.131:9092", "192.168.59.131:9093", "192.168.59.131:9094"),// NOTE: When Topic is not defined here, each Message must define it instead.// 注意：如果这里没有定义主题，则每条信息都必须定义主题Balancer: &kafka.LeastBytes{},}err := w.WriteMessages(context.Background(),// NOTE: Each Message has Topic defined, otherwise an error is returned.// 注意：每条信息都定义了主题，否则将返回错误信息。kafka.Message{Topic: "topic-A",Key:   []byte("Key-A"),Value: []byte("Hello World!"),},kafka.Message{Topic: "topic-B",Key:   []byte("Key-B"),Value: []byte("One!"),},kafka.Message{Topic: "topic-C",Key:   []byte("Key-C"),Value: []byte("Two!"),},)if err != nil {log.Fatal("failed to write messages:", err)}if err := w.Close(); err != nil {log.Fatal("failed to close writer:", err)}
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/54995.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

10.Oracle中decode函数

10.Oracle中decode函数

【函数格式】&#xff1a; decode ( expression, condition_01, result_01, condition_02, result_02, ......, condition_n, result_n, result_default) 【函数说明】&#xff1a; 若表达式expression值与condition_01值匹配&#xff0c;则返回result_01&#xff0c;…
阅读更多...
CPU深度解析

CPU深度解析

操作系统课程 计算机组成 ALU：计算单元（运算器）PC：pc寄存器存执行指令Registers：寄存器存数据MMU：控制器程序的构成：指令+数据 总线：一个程序读入内存，全是由0和1构成，从内存读取到cpu计算，需要通过总线。一段01数据段是指令还是数据是通过来源总线区分的。总线分…
阅读更多...
flutter和原生利用pigeon建立通道

flutter和原生利用pigeon建立通道

首先导入依赖&#xff1a; dependencies: pigeon: ^10.0.0定义一个文件&#xff1a; /// 用于定于flutter和平台的桥接方法 /// HostApi() 标记的&#xff0c;是用于 Flutter 调用原生的方法&#xff1b; /// FlutterApi() 标记的&#xff0c;是用于原生调用 Flutter 的方法&…
阅读更多...
Netty入门学习和技术实践

Netty入门学习和技术实践

Netty入门学习和技术实践 Netty1.Netty简介2.IO模型3.Netty框架介绍4. Netty实战项目学习5. Netty实际应用场景6.扩展 Netty 1.Netty简介 Netty是由JBOSS提供的一个java开源框架&#xff0c;现为 Github上的独立项目。Netty提供异步的、事件驱动的网络应用程序框架和工具&…
阅读更多...
高精度地图定位在高速公路自动驾驶系统中的应用

高精度地图定位在高速公路自动驾驶系统中的应用

近年来随着汽车保有量不断增加&#xff0c;随之而来的是: ( 1) 严重的交通拥堵&#xff0c;通行效率低下&#xff0c;用在通行上的时间不断增加; ( 2) 交通事故频发&#xff0c;交通事故导致的伤亡人数和费用不断增加&#xff0c;而且绝大多数事故是由人为因素导致的; ( 3) 大气…
阅读更多...
2023年6月电子学会Python等级考试试卷（一级）答案解析

2023年6月电子学会Python等级考试试卷（一级）答案解析

青少年软件编程（Python）等级考试试卷（一级） 一、单选题(共25题，共50分) 1. 可以对Python代码进行多行注释的是？（　） A. #
阅读更多...
leetcode-242.有效的字母异位词-day20

leetcode-242.有效的字母异位词-day20

阅读更多...
Linux：Nginx服务与搭建

Linux：Nginx服务与搭建

目录 一、Nginx概述 二、Nginx三大作用&#xff1a;反向代理、负载均衡、动静分离 三、Nginx和Apache 3.1Nginx和Apache的差异 3.2Nginx和Apache的优缺点比较 四、编译安装niginx 五、创建Nginx 自启动文件 六、Nginx的信号使用 6.1信号 七、升级 nginx1.18 nginx1.2…
阅读更多...
题解：ABC281C - Circular Playlist

题解：ABC281C - Circular Playlist

题解&#xff1a;ABC281C - Circular Playlist 题目 链接&#xff1a;Atcoder。 链接&#xff1a;洛谷。 难度 算法难度&#xff1a;C。 思维难度&#xff1a;C。 调码难度&#xff1a;C。 综合评价&#xff1a;入门。 算法 模拟循环取模优化。 思路 记从Song1至So…
阅读更多...
企业展示小程序的制作流程及关键步骤详解

企业展示小程序的制作流程及关键步骤详解

在移动互联网时代&#xff0c;企业展示小程序已成为各个行业推广和展示的重要工具。搭建一个企业展示小程序不仅能够提高企业形象&#xff0c;还能够增加用户粘性和提升用户体验。下面我们来看一下如何从零基础搭建一个企业展示小程序&#xff0c;并顺利上线。 第一步&#xff…
阅读更多...
2023.8 - java - 多态

2023.8 - java - 多态

多态是同一个行为具有多个不同表现形式或形态的能力。 多态就是同一个接口&#xff0c;使用不同的实例而执行不同操作&#xff0c; 多态的优点 1. 可替换性2 可扩充性3. 接口性、灵活性、简化性4. 消除类型之间的耦合关系 多态存在的三个必要条件 继承重写父类引用指向子类…
阅读更多...
代码随想录第五十天

代码随想录第五十天

代码随想录第五十天 Leetcode 123. 买卖股票的最佳时机 IIILeetcode 188. 买卖股票的最佳时机 IV Leetcode 123. 买卖股票的最佳时机 III 题目链接: 买卖股票的最佳时机 III 自己的思路:想不到&#xff01;&#xff01;&#xff01;&#xff01;高维dp数组&#xff01;&#x…
阅读更多...
每日一题：leetcode 56 合并区间

每日一题：leetcode 56 合并区间

以数组 intervals 表示若干个区间的集合&#xff0c;其中单个区间为 intervals[i] [starti, endi] 。请你合并所有重叠的区间&#xff0c;并返回 一个不重叠的区间数组&#xff0c;该数组需恰好覆盖输入中的所有区间 。 示例 1&#xff1a; 输入&#xff1a;intervals [[1,…
阅读更多...
【负载均衡】Nacos简单入门

【负载均衡】Nacos简单入门

Nacos简单入门 快速安装 在Nacos的GitHub页面&#xff0c;提供有下载链接&#xff0c;可以下载编译好的Nacos服务端或者源代码&#xff1a; 下载完压缩包之后&#xff0c;放在任意目录下面进行解压&#xff1a; GitHub主页&#xff1a;https://github.com/alibaba/nacos G…
阅读更多...
Android源码——从Looper看ThreadLocal

Android源码——从Looper看ThreadLocal

1 概述 ThreadLocal用于在当前线程中存储数据&#xff0c;由于存储的数据只能在当前线程内使用&#xff0c;所以自然是线程安全的。 Handler体系中&#xff0c;Looper只会存在一个实例&#xff0c;且只在当前线程使用&#xff0c;所以使用ThreadLocal进行存储。 2 存储原理 …
阅读更多...
机器学习策略——优化深度学习系统

机器学习策略——优化深度学习系统

正交化&#xff08;Orthogonalization&#xff09; 老式电视机&#xff0c;有很多旋钮可以用来调整图像的各种性质&#xff0c;对于这些旧式电视&#xff0c;可能有一个旋钮用来调图像垂直方向的高度&#xff0c;另外有一个旋钮用来调图像宽度&#xff0c;也许还有一个旋钮用来…
阅读更多...
AIGC ChatGPT 按年份进行动态选择的动态图表

AIGC ChatGPT 按年份进行动态选择的动态图表

动态可视化分析的好处与优势&#xff1a; 1. 提高信息理解性&#xff1a;可视化分析使得大量复杂的数据变得易于理解&#xff0c;通过图表、颜色、形状、尺寸等方式&#xff0c;能够直观地表现不同的数据关系和模式。 2. 加快决策速度&#xff1a;数据可视化可以帮助用户更快…
阅读更多...
04_19linux自己撸内存池实战,仿造slab分配器

04_19linux自己撸内存池实战,仿造slab分配器

前言 自己撸一个内存池 其实就相当于linux里面带的 slab分配器 可以翻翻之前的章 看看slab 和伙伴分配器的不同 在学习c语言时&#xff0c;我们常常会使用到malloc()去申请一块内存空间&#xff0c;用于存放我们的数据。刚开始我们只要知道申请内存时使用用malloc去申请一块就…
阅读更多...
RV64函数调用流程分析

RV64函数调用流程分析

RV64函数调用流程分析 1 RV64 函数调用实例2 对应代码的分析2.1 main函数及其对应的汇编程序2.1.1 main的C代码实现2.1.2 main函数对应汇编及其分析2.1.3 执行完成之后栈的存放情况 2.2 test_fun_a函数及其对应的汇编程序2.2.1 test_fun_a函数的C实现2.2.2 test_fun_a函数对应汇…
阅读更多...
C# 与 .NET 之间的关系和区别

C# 与 .NET 之间的关系和区别

C# 语言介绍 C#&#xff08;读作“See Sharp”&#xff09;是一种新式编程语言&#xff0c;不仅面向对象&#xff0c;还类型安全。 开发人员利用 C# 能够生成在 .NET 中运行的多种安全可靠的应用程序。 C# 源于 C 语言系列&#xff0c;C、C、Java 和 JavaScript 程序员很快就可…
阅读更多...
最新文章

Copyright @ 2022~2023