介绍

如果您需要了解 Kafka 的基础知识，例如它的主要功能、组件和优势，我在这里有一篇文章对此进行了介绍。请查看它并按照步骤操作，直到使用 Docker 完成 Kafka 安装，然后继续以下部分。

使用 Golang 连接到 Kafka

与Kafka与NodeJS连接的文章中的示例类似，该源码也包括两部分：初始化一个生产者 发送 消息 到 Kafka 并使用 消费者 订阅来自 [ 的消息&&&]话题。

我将把代码分解成更小的部分以便更好地理解。首先，让我们定义变量值。

包主 进口 （ “FMMT” “github.com/confluenceinc/confluence-kafka-go/kafka” ） 变量（ 经纪人=“本地主机：9092” groupId = "组 ID" 主题 =“主题名称” ）

package main

import (
  "fmt"
  "github.com/confluentinc/confluent-kafka-go/kafka"
)

var (
  broker  = "localhost:9092"
  groupId = "group-id"
  topic   = "topic-name"
)

-这里，包

github.com/confluenceinc/confluence-kafka-go/kafka用于连接到Kafka。

-

经纪商是主机地址；如果您使用ZooKeeper，请相应地替换主机地址。

-

groupId和topic可以根据需要更改。

接下来是初始化生产者。

func startProducer() { p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": 代理}) 如果错误！= nil { 恐慌（错误） } 去函数（）{ for e := 范围 p.Events() { 开关 ev := e.(类型) { 案例*kafka.消息： 如果 ev.TopicPartition.Error != nil { fmt.Printf("传送失败：%v\n", ev.TopicPartition) } 别的 { fmt.Printf("已将消息发送至 %v\n", ev.TopicPartition) } } } }() for _, word := range []string{"消息 1", "消息 2", "消息 3"} { p.Produce(&kafka.Message{ TopicPartition: kafka.TopicPartition{主题: &topic, 分区: kafka.PartitionAny}, 值：[]字节（字）， }，无） } }

package main

import (
  "fmt"
  "github.com/confluentinc/confluent-kafka-go/kafka"
)

var (
  broker  = "localhost:9092"
  groupId = "group-id"
  topic   = "topic-name"
)

上面的代码用于将消息数组

{"message 1", "message 2", "message 3"} 发送到主题并使用 go-routine 用 for e := range p.Events() 迭代事件并打印出传递结果，无论是成功或失败。

下一步是创建一个

消费者来订阅到主题并接收消息。

func startConsumer() { c, err := kafka.NewConsumer(&kafka.ConfigMap{ “bootstrap.servers”：经纪人， "group.id": 组ID, "auto.offset.reset": "最早", }) 如果错误！= nil { 恐慌（错误） } c.订阅（主题，nil） 为了 { 消息，错误：= c.ReadMessage(-1) 如果错误==零{ fmt.Printf("%s 上的消息：%s\n", msg.TopicPartition, string(msg.Value)) } 别的 { fmt.Printf("消费者错误: %v (%v)\n", err, msg) 休息 } } c.关闭() }

package main

import (
  "fmt"
  "github.com/confluentinc/confluent-kafka-go/kafka"
)

var (
  broker  = "localhost:9092"
  groupId = "group-id"
  topic   = "topic-name"
)

最后，由于这是一个简单的示例，因此调用函数创建

生产者和消费者以供使用。在现实场景中，生产者和消费者的部署通常在微服务系统中的两个不同服务器上完成。

func main() { 启动生产者() 启动消费者() }

package main

import (
  "fmt"
  "github.com/confluentinc/confluent-kafka-go/kafka"
)

var (
  broker  = "localhost:9092"
  groupId = "group-id"
  topic   = "topic-name"
)

编码愉快！

---

如果您觉得本文内容有帮助，请访问我博客上的原文，支持作者，探索更多有趣的内容。

---

您可能会感兴趣的一些系列：

NodeJS
 反应
Docker 
Kubernetes