用golang实现kafka

时间：2023-05-10 17:30

随着企业级应用程序架构的逐渐复杂化，消息传输变成了一个至关重要的组成部分。这时Kafka便崭露头角。Kafka是一款高效可靠的分布式消息队列，它支持消息的发布和订阅，是现代化的企业级消息系统，且拥有非常高的吞吐量和低延迟。在Kafka的API中，尽管官方提供了多种语言的客户端，但近年来Golang的应用越来越广泛，所以本文以Golang作为实现语言，讲解如何用Golang实现Kafka。

一、依赖

在开始之前，需要先下载所需的依赖：

sarama：Golang Kafka客户端库
pkg/errors：对Go标准库的错误包进行封装

具体使用方法如下：

go get github.com/Shopify/sarama
go get github.com/pkg/errors

二、创建一个生产者

在介绍Kafka的API之前，需要先创建一个生产者实例。生产者的代码如下所示：

package mainimport (    "fmt"    "time"    "github.com/pkg/errors"    "github.com/Shopify/sarama")func main() {    config := sarama.NewConfig()    config.Producer.Partitioner = sarama.NewRandomPartitioner    config.Producer.RequiredAcks = sarama.WaitForAll    config.Producer.Return.Successes = true    producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)    if err != nil {        panic(errors.Wrap(err, "failed to create producer"))    }    defer producer.Close()    for i := 0; i < 10; i++ {        message := &sarama.ProducerMessage{            Topic: "test_topic",            Value: sarama.StringEncoder(fmt.Sprintf("test message %d", i)),        }        partition, offset, err := producer.SendMessage(message)        if err != nil {            fmt.Println(errors.Wrapf(err, "failed to send message: %s", message))        } else {            fmt.Printf("message sent to partition %d at offset %d", partition, offset)        }        time.Sleep(500 * time.Millisecond) // 延迟发送    }}

代码中主要做了以下几件事情：

配置生产者：设置生产者的配置，并指定分区方式为随机分区，要求等待所有ISR节点都确认消息后再返回，以及在发送成功后返回Partition和Offset。
创建生产者：用指定的broker地址和配置创建一个生产者实例。
发送消息：创建一个含有消息主题和内容的消息，并进行发送。
输出结果：打印结果，记录消息分区和偏移量。

三、创建一个消费者

在其次，需要创建一个消费者实例。消费者的代码如下所示：

package mainimport (    "context"    "fmt"    "os"    "os/signal"    "github.com/Shopify/sarama"    "github.com/pkg/errors")func main() {    config := sarama.NewConfig()    config.Consumer.Return.Errors = true    consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, config)    if err != nil {        panic(errors.Wrap(err, "failed to create consumer"))    }    defer consumer.Close()    signals := make(chan os.Signal, 1)    signal.Notify(signals, os.Interrupt)    partitions, err := consumer.Partitions("test_topic")    if err != nil {        panic(errors.Wrapf(err, "failed to read partitions for topic: test_topic"))    }    ctx, cancel := context.WithCancel(context.Background())    for _, partition := range partitions {        go func(partition int32) {            partitionConsumer, err := consumer.ConsumePartition("test_topic", partition, sarama.OffsetOldest)            if err != nil {                fmt.Printf("failed to create partition consumer for partition %d: %s", partition, err)                return            }            defer partitionConsumer.Close()            for {                select {                case msg := <-partitionConsumer.Messages():                    fmt.Printf("Consumed message from partition %d at offset %d: %s", msg.Partition, msg.Offset, msg.Value)                case <-signals:                    cancel()                    return                case err := <-partitionConsumer.Errors():                    fmt.Printf("Consumed error from partition %d: %s", partition, err)                case <-ctx.Done():                    return                }            }        }(partition)    }    <-signals    fmt.Println("Shutting down consumer")}

代码中主要做了以下几件事情：