golang连接kafka的示例代码

接下来我将详细讲解如何连接Kafka并使用Golang进行消息传输的完整攻略，其中包含两个示例说明。

环境准备

在开始之前，需要确保已经安装好以下工具：

• 一个Kafka服务
• Golang的开发环境

在控制台中执行以下命令安装Kafka依赖：

$ go get github.com/segmentio/kafka-go

在本示例中，我们将使用github.com/segmentio/kafka-go包来连接和发送消息给Kafka。

示例1：连接Kafka

package main

import (
    "context"
    "fmt"

    kafka "github.com/segmentio/kafka-go"
)

func main() {
    // 定义Kafka主题和服务地址
    topic := "my-topic"
    broker := []string{"kafka1:9092", "kafka2:9092", "kafka3:9092"}

    // 配置Kafka连接
    conf := kafka.ReaderConfig{
        Brokers:   broker,
        Topic:     topic,
        Partition: 0,
        MaxBytes:  10e6,
    }

    // 创建Kafka reader
    reader := kafka.NewReader(conf)

    defer func() {
        if err := reader.Close(); err != nil {
            fmt.Println("Failed to close reader:", err)
        }
    }()

    // 循环从Kafka读取消息
    for {
        msg, err := reader.ReadMessage(context.Background())
        if err != nil {
            fmt.Println("Failed to read message:", err)
            break
        }

        fmt.Printf("Received message: %v\n", string(msg.Value))
    }
}

上述代码连接了Kafka，并读取了一个主题中的消息。以阻塞方式循环从Kafka读取消息，并打印出消息。

示例2：向Kafka写入消息

package main

import (
    "context"
    "fmt"
    "time"

    kafka "github.com/segmentio/kafka-go"
)

func main() {
    // 定义Kafka主题和服务地址
    topic := "my-topic"
    broker := []string{"kafka1:9092", "kafka2:9092", "kafka3:9092"}

    // 配置Kafka连接
    conf := kafka.WriterConfig{
        Brokers:  broker,
        Topic:    topic,
        Balancer: &kafka.LeastBytes{},
    }

    // 创建Kafka writer
    writer := kafka.NewWriter(conf)

    defer func() {
        if err := writer.Close(); err != nil {
            fmt.Println("Failed to close writer:", err)
        }
    }()

    // 定义消息
    message := kafka.Message{
        Value: []byte(fmt.Sprintf("Hello from Kafka at %v!", time.Now().Format(time.RFC3339))),
    }

    // 发送消息
    err := writer.WriteMessages(context.Background(), message)
    if err != nil {
        fmt.Println("Failed to write message:", err)
    } else {
        fmt.Println("Message written to Kafka!")
    }
}

上述代码演示了如何向Kafka发送消息。

首先定义了Kafka主题和服务地址，然后进行了连接配置。然后定义了要发送的消息内容，并向Kafka发送了一条消息。

请注意，这里使用的是kafka.NewWriter()方法可以让你配置更多与写入有关的参数。

这里我们使用了随机的kafka.LeastBytes负载均衡器将消息发送到分区。发送消息时，我们调用了writer.WriteMessages()方法传递消息和上下文对象。

总结

以上就是连接Kafka并使用Golang发送和接收消息的完整攻略。如果你有任何问题或建议，请在评论区留言。