1. Toy模板网首页
  2. > Toy博客

go-kafka

9月前 作者：哒哒-blog 分类：Toy博客 阅读(29) 违法举报

这篇具有很好参考价值的文章主要介绍了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 main

import (
	"context"
	"github.com/segmentio/kafka-go"
	"log"
	"time"
)

func main() {
	// to produce messages
	topic := "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 main

import (
	"context"
	"fmt"
	"github.com/segmentio/kafka-go"
	"log"
	"time"
)

func main() {
	// to consume messages
	topic := "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.SetReadDeadline(time.Now().Add(10 * time.Second))
	batch := conn.ReadBatch(10e3, 1e6) // fetch 10KB min, 1MB max

	b := make([]byte, 10e3) // 10KB max per message
	for {
		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 main

import (
	"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.Conn
	controllerConn, 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 main

import (
	"context"
	"fmt"
	"github.com/segmentio/kafka-go"
	"log"
)

// kafka
func main() {
	// make a new reader that consumes from topic-A, partition 0, at offset 42
	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, // 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 main

import (
	"context"
	"fmt"
	"github.com/segmentio/kafka-go"
	"log"
)

// kafka
func main() {
	// make a new reader that consumes from topic-A, partition 0, at offset 42
	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, // 10MB
		GroupID:   "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 main

import (
	"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, // 10MB
		GroupID:   "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, // 10MB
		GroupID:        "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 main

import (
	"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 main

import (
	"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 error
	const retries = 3
	for 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 来按消息定义主题。文章来源地址https://www.toymoban.com/news/detail-667955.html

package main

import (
	"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)
	}
}

到了这里，关于go-kafka的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处： 如若内容造成侵权/违法违规/事实不符，请点击违法举报进行投诉反馈，一经查实，立即删除！

分享到：
领支付宝红包 赞助服务器费用

相关文章

  • Go语言（Golang）数据库编程

    要想连接到 SQL 数据库，首先需要加载目标数据库的驱动，驱动里面包含着于该数据库交互的逻辑。 sql.Open() 数据库驱动的名称 数据源名称 得到一个指向 sql.DB 这个 struct 的指针 sql.DB 是用来操作数据库的，它代表了0个或者多个底层连接的池，这些连接由sql 包来维护，sql 包会

    2024年02月03日
    浏览(93)
  • 【Golang】VScode配置Go语言环境

    【Golang】VScode配置Go语言环境

    安装VScode请参考我的上一篇博客：VScode安装_㫪548的博客-CSDN博客 接下来我们直接进入正题： Go语言（又称Golang）是一种开源的编程语言，由Google开发并于2009年首次发布。Go语言具有简洁、高效、可靠和易于阅读的特点，被设计用于解决大型项目的开发需求。它结合了静态类型

    2024年02月03日
    浏览(66)

  • 【Golang】Golang进阶系列教程--Go 语言 map 如何顺序读取？

    Go 语言中的 map 是一种非常强大的数据结构，它允许我们快速地存储和检索键值对。 然而，当我们遍历 map 时，会有一个有趣的现象，那就是输出的键值对顺序是不确定的。 先看一段代码示例： 当我们多执行几次这段代码时，就会发现，输出的顺序是不同的。 首先，Go 语言

    2024年02月14日
    浏览(69)
  • 【Golang】Golang进阶系列教程--Go 语言数组和切片的区别

    【Golang】Golang进阶系列教程--Go 语言数组和切片的区别

    在 Go 语言中，数组和切片看起来很像，但其实它们又有很多的不同之处，这篇文章就来说说它们到底有哪些不同。 数组和切片是两个常用的数据结构。它们都可以用于存储一组相同类型的元素，但在底层实现和使用方式上存在一些重要的区别。 Go 中数组的长度是不可改变的

    2024年02月15日
    浏览(61)
  • 【Golang】Golang进阶系列教程--Go 语言切片是如何扩容的？

    【Golang】Golang进阶系列教程--Go 语言切片是如何扩容的？

    在 Go 语言中，有一个很常用的数据结构，那就是切片（Slice）。 切片是一个拥有相同类型元素的可变长度的序列，它是基于数组类型做的一层封装。它非常灵活，支持自动扩容。 切片是一种引用类型，它有三个属性：指针，长度和容量。 底层源码定义如下： 指针： 指向

    2024年02月14日
    浏览(70)
  • 【Golang】三分钟让你快速了解Go语言&为什么我们需要Go语言？

    【Golang】三分钟让你快速了解Go语言&为什么我们需要Go语言？

    博主简介： 努力学习的大一在校计算机专业学生，热爱学习和创作。目前在学习和分享：数据结构、Go，Java等相关知识。 博主主页： @是瑶瑶子啦 所属专栏: Go语言核心编程 近期目标： 写好专栏的每一篇文章 Go 语言从 2009 年 9 月 21 日开始作为谷歌公司 20% 兼职项目，即相关

    2023年04月21日
    浏览(62)
  • Golang（Go语言）IP地址转换函数

    Golang（Go语言）IP地址转换函数

    String形式的IP地址和Int类型互转函数 代码 输出如下：  

    2024年02月05日
    浏览(54)
  • Golang区块链钱包_go语言钱包

    Golang区块链钱包_go语言钱包

    Golang区块链钱包的特点 Golang区块链钱包具有以下几个特点： 1. 高性能 Golang是一种编译型语言，具有快速的执行速度和较低的内存消耗。这使得Golang区块链钱包在处理大规模交易数据时表现出色，能够满足高性能的需求。 2. 并发支持 Golang内置了轻量级线程——goroutine，以及

    2024年04月15日
    浏览(62)
  • 【GoLang】哪些大公司正在使用Go语言

    【GoLang】哪些大公司正在使用Go语言

    前言： 随着计算机科学和软件开发的快速发展，编程语言的选择变得愈加关键。 在这个多元化的编程语境中，Go语言（简称Golang）以其简洁、高效、并发处理能力等特性逐渐受到业界关注。 越来越多的大型科技公司纷纷采用Go语言作为其软件开发的首选语言，这种趋势反映了

    2024年02月04日
    浏览(65)

  • 1分钟带你了解golang(go语言)

    Golang：也被称为Go语言，是一种开源的编程语言。由Google的Robert Griesemer、Rob Pike和Ken Thompson于2007年开始设计，2009年11月正式对外发布。（被誉为21世纪的C语言） 像python一样的优雅，有c一样的性能 天生的协程 编译快 … 编辑器： goland（推荐） 下载链接 vscode 官网：golang.goog

    2024年02月03日
    浏览(43)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包，优惠每天领

二维码1

领取红包

二维码2

领红包