做效果图常用的网站,做外账要登什么网站,网站建设兰州,网站开发软件学习文章目录 Apache kafka简介开始使用Apache Kafka构建生产者构建消费者 总结 之前已经有两篇文章介绍过
Go如何操作
kafka 28.windows安装kafka#xff0c;Go操作kafka示例#xff08;sarama库#xff09; 51.Go操作kafka示例#xff08;kafka-go库#xff09; Apache ka… 文章目录 Apache kafka简介开始使用Apache Kafka构建生产者构建消费者 总结 之前已经有两篇文章介绍过
Go如何操作
kafka 28.windows安装kafkaGo操作kafka示例sarama库 51.Go操作kafka示例kafka-go库 Apache kafka简介 Apache Kafka是一个开源分布式事件流平台用于高性能数据管道、流式分析、数据集成和关键任务应用程序。它提供了诸如流数据管道、实时分析、数据集成、事件源和日志聚合等多种能力。
Kafka的用例和能力 流数据管道 Kafka提供了一个分布式发布-订阅消息系统可以在系统或应用程序之间流式传输数据。它提供了具有数据复制和容错能力的强大队列。 实时分析Kafka允许使用工具如Kafka Streams和KSQL处理实时数据流用于构建流式分析和数据处理应用程序。 数据集成 Kafka可以用来通过在不同数据源和格式之间流式传输数据来集成不同的系统。这使它对流式ETL非常有用。 事件源 Kafka提供了可以重放的事件时间日志用于重构应用程序状态适用于事件源和CQRS模式。 日志聚合 Kafka通常用于将不同服务器和应用程序的日志聚合到一个中央存储库中。这允许统一访问日志数据。
为什么将Golang与Apache Kafka结合使用
将Golang这一高效并发的编程语言与Apache Kafka这一分布式事件流平台结合起来提供了一个在构建尖端现代应用程序方面表现出色的强大技术栈。这两种技术之间的协同作用源自几个关键优势 性能 : Golang和Apache Kafka都提供高性能。Golang快速、高效和轻量级。Kafka为速度而构建具有高吞吐量和低延迟。它们一起可以处理苛刻的工作负载。 可扩展性 : Golang的goroutines和Kafka的分区允许应用程序水平扩展以处理大量数据。Kafka可以轻松扩展生产者和消费者。 并发性 : Golang通过goroutines和channels提供了出色的并发编程能力。Kafka并发传递消息并支持并行性。 可用性 : Kafka的分布式架构使其高度可用和容错。Golang应用可以利用这一点来构建弹性系统。 互操作性 : Kafka有多种语言的客户端允许Golang应用与多语言环境互动。Kafka还使用二进制TCP协议以提高效率。 现代设计 : Kafka和Golang都采用现代设计理念使它们非常适合云原生和微服务架构。 开发人员体验 : Kafka的客户端库结合Goroutines、channels和接口使其易于使用。
Kafka和Golang将性能、可扩展性和并发与生产力结合在一起 - 使它们成为构建可扩展的服务、管道和流应用程序的绝佳选择。
开始使用Apache Kafka
在开始使用Golang和Apache Kafka之前我们必须确保golang和Kafka已经安装并在我们的机器上运行。
安装Kafka 28.windows安装kafkaGo操作kafka示例sarama库
Apache Kafka的Golang包 您可以使用go get安装confluent-kafka-go包
go get -u github.com/confluentinc/confluent-kafka-go/kafka安装后您可以在Go代码中导入并使用confluent-kafka-go。
package mainimport (fmtgithub.com/confluentinc/confluent-kafka-go/kafka
)func main() {p, err : kafka.NewProducer(kafka.ConfigMap{bootstrap.servers: localhost:9092})if err ! nil {fmt.Printf(创建生产者失败: %s\n, err)return}// 生产消息到主题处理交付报告等。// 使用后记得关闭生产者defer p.Close()
}构建生产者
Kafka生产者是Apache Kafka生态系统中的一个关键组成部分作为一个客户端应用程序负责向Kafka集群发布写入事件。这一部分提供了关于Kafka生产者的全面概述以及针对调整其行为的配置设置的初步探讨。
下面是一个Golang应用程序的示例它生产数据并将其发布到Kafka的具体topic。它还说明了如何在Golang中为Kafka消息序列化数据并演示了如何处理错误和重试。
package mainimport (fmtgithub.com/confluentinc/confluent-kafka-go/kafka
)const (kafkaBroker localhost:9092topic test-topic
)type Messagestruct {Key string json:keyValue string json:value
}func main() {// 创建一个新的Kafka生产者p, err : kafka.NewProducer(kafka.ConfigMap{bootstrap.servers: kafkaBroker})if err ! nil {fmt.Printf(创建生产者失败: %s\n, err)return}defer p.Close()// 定义要发送的消息message : Message{Key: example_key,Value: Hello, Kafka!,}// 序列化消息serializedMessage, err : serializeMessage(message)if err ! nil {fmt.Printf(消息序列化失败: %s\n, err)return}// 将消息生产到Kafka主题err produceMessage(p, topic, serializedMessage)if err ! nil {fmt.Printf(消息生产失败: %s\n, err)return}fmt.Println(消息成功生产)
}func serializeMessage(message Message) ([]byte, error) {// 将消息结构体序列化为JSONserialized, err : json.Marshal(message)if err ! nil {return nil, fmt.Errorf(消息序列化失败: %w, err)}return serialized, nil
}func produceMessage(p *kafka.Producer, topic string, message []byte) error {// 创建一个新的要生产的Kafka消息kafkaMessage : kafka.Message{TopicPartition: kafka.TopicPartition{Topic: topic, Partition: kafka.PartitionAny},Value: message,}// 生产Kafka消息deliveryChan : make(chan kafka.Event)err : p.Produce(kafkaMessage, deliveryChan)if err ! nil {return fmt.Errorf(消息生产失败: %w, err)}// 等待交付报告或错误e : -deliveryChanm : e.(*kafka.Message)// 检查交付错误即生成者方确保发送到Broker的消息不丢失// 但可能重复发送如实际发成功了但是Broker的ACK返回给生产者时出现了网络Error// 从而重试后导致消息重复发送这时候需要下游做好幂等处理if m.TopicPartition.Error ! nil {return fmt.Errorf(交付失败: %s, m.TopicPartition.Error)}// 关闭交付频道close(deliveryChan)return nil
}步骤解释 创建一个Kafka生产者。 使用json.Marshal函数将自定义消息结构体Message序列化为JSON。 使用生产者将序列化的消息生产到Kafka topic。 使用交付报告和错误检查处理错误和重试。
确保将localhost:9092替换为您的Kafka代理地址将test-topic替换为所需的主题名称。此外您可能需要处理更复杂的错误场景并根据您的具体需求实现重试逻辑。
构建消费者
Kafka消费者就像小型事件处理器它们获取并消化数据流。它们订阅主题并消费任何新到达的消息处理每一个消息。我们将探讨这些消费者的内部工作原理和调整其性能的配置。
下面是一个Golang应用程序的示例它从Kafka主题消费消息。如下代码包括了如何处理消费到的消息的说明以及对不同消费模式如单个消费者和消费者组的讨论。
package mainimport (fmtosos/signalgithub.com/confluentinc/confluent-kafka-go/kafka
)const (kafkaBroker localhost:9092topic test-topicgroupID test-group
)func main() {// 创建一个新的Kafka消费者c, err : kafka.NewConsumer(kafka.ConfigMap{bootstrap.servers: kafkaBroker,group.id: groupID, // 消费者组标识auto.offset.reset: earliest, // 从头开始消费})if err ! nil {fmt.Printf(创建消费者失败: %s\n, err)return}defer c.Close()// 订阅Kafka主题err c.SubscribeTopics([]string{topic}, nil)if err ! nil {fmt.Printf(订阅主题失败: %s\n, err)return}// 设置一个通道来处理操作系统信号以便优雅地关闭sigchan : make(chan os.Signal, 1)signal.Notify(sigchan, os.Interrupt)// 开始消费消息run : truefor run true {select {case sig : -sigchan:fmt.Printf(接收到信号 %v: 正在终止\n, sig)run falsedefault:// 轮询Kafka消息1次最多拉取100条消息ev : c.Poll(100) if ev nil {continue}switch e : ev.(type) {case *kafka.Message:// 处理消费的消息fmt.Printf(从主题 %s 收到消息: %s\n, *e.TopicPartition.Topic, string(e.Value))case kafka.Error:// 处理Kafka错误fmt.Printf(错误: %v\n, e)}}}
}步骤解释 创建一个Kafka消费者。 订阅一个Kafka主题。 设置一个通道来处理操作系统信号如SIGINT以优雅地关闭。 开始从订阅的Topic消费消息。 处理消费的消息以及Kafka错误。
不同的消费模式 单个消费者在这种模式下单个消费者实例从主题的一个或多个分区读取消息。当您只需要一个消费者应用程序实例来处理来自Topic的所有消息时这很有用。 消费者组消费者组允许您通过将消息处理分布到多个消费者实例来扩展消费以实现扩展。每个消费者组可以有多个消费者组内的每个消费者从一部分分区读取消息。这使得消息的并行处理成为可能提供了容错能力和高吞吐量。
在提供的示例中group.id配置设置用于指定消费者组ID。这允许消费者应用程序的多个实例在消费者组中一起工作从Kafka Topic消费消息。
总结
总之Apache Kafka作为构建实时数据管道和流应用程序的强大解决方案得益于其分布式、可扩展和容错的架构。当与Golang结合时它形成了一个在性能、可扩展性和并发方面表现出色的强大技术栈非常适合现代应用程序。通过利用Kafka的功能和Golang的优势开发人员可以构建出具有弹性和高性能的服务、管道和流应用程序这些应用程序可以轻松扩展以满足当今数据驱动世界的需求。无论是处理实时分析、集成不同的系统还是聚合日志Kafka和Golang提供了一个强势组合使开发人员能够轻松构建创新和可扩展的解决方案。 文章转载自： http://www.morning.lqpzb.cn.gov.cn.lqpzb.cn http://www.morning.okiner.com.gov.cn.okiner.com http://www.morning.sqfrg.cn.gov.cn.sqfrg.cn http://www.morning.rkqzx.cn.gov.cn.rkqzx.cn http://www.morning.mrccd.cn.gov.cn.mrccd.cn http://www.morning.bxqpl.cn.gov.cn.bxqpl.cn http://www.morning.gccdr.cn.gov.cn.gccdr.cn http://www.morning.srbmc.cn.gov.cn.srbmc.cn http://www.morning.xhhzn.cn.gov.cn.xhhzn.cn http://www.morning.kflbf.cn.gov.cn.kflbf.cn http://www.morning.tqrbl.cn.gov.cn.tqrbl.cn http://www.morning.hnrls.cn.gov.cn.hnrls.cn http://www.morning.qkzdc.cn.gov.cn.qkzdc.cn http://www.morning.bdsyu.cn.gov.cn.bdsyu.cn http://www.morning.pabxcp.com.gov.cn.pabxcp.com http://www.morning.mnyzz.cn.gov.cn.mnyzz.cn http://www.morning.xlndf.cn.gov.cn.xlndf.cn http://www.morning.srzhm.cn.gov.cn.srzhm.cn http://www.morning.lfqtp.cn.gov.cn.lfqtp.cn http://www.morning.ygwyt.cn.gov.cn.ygwyt.cn http://www.morning.lwmzp.cn.gov.cn.lwmzp.cn http://www.morning.cwzzr.cn.gov.cn.cwzzr.cn http://www.morning.mszwg.cn.gov.cn.mszwg.cn http://www.morning.kmbgl.cn.gov.cn.kmbgl.cn http://www.morning.wjqbr.cn.gov.cn.wjqbr.cn http://www.morning.smspc.cn.gov.cn.smspc.cn http://www.morning.tqlhn.cn.gov.cn.tqlhn.cn http://www.morning.zhengdaotang.cn.gov.cn.zhengdaotang.cn http://www.morning.jcxzq.cn.gov.cn.jcxzq.cn http://www.morning.lbbgf.cn.gov.cn.lbbgf.cn http://www.morning.hxgly.cn.gov.cn.hxgly.cn http://www.morning.wfmqc.cn.gov.cn.wfmqc.cn http://www.morning.rbnnq.cn.gov.cn.rbnnq.cn http://www.morning.stxg.cn.gov.cn.stxg.cn http://www.morning.pqyms.cn.gov.cn.pqyms.cn http://www.morning.kpxky.cn.gov.cn.kpxky.cn http://www.morning.weiwt.com.gov.cn.weiwt.com http://www.morning.nbgfz.cn.gov.cn.nbgfz.cn http://www.morning.nqrlz.cn.gov.cn.nqrlz.cn http://www.morning.xnkh.cn.gov.cn.xnkh.cn http://www.morning.rjfr.cn.gov.cn.rjfr.cn http://www.morning.pwbps.cn.gov.cn.pwbps.cn http://www.morning.jkbqs.cn.gov.cn.jkbqs.cn http://www.morning.cjnfb.cn.gov.cn.cjnfb.cn http://www.morning.zxqqx.cn.gov.cn.zxqqx.cn http://www.morning.rnds.cn.gov.cn.rnds.cn http://www.morning.tbjtm.cn.gov.cn.tbjtm.cn http://www.morning.spghj.cn.gov.cn.spghj.cn http://www.morning.pmdzd.cn.gov.cn.pmdzd.cn http://www.morning.rxsgk.cn.gov.cn.rxsgk.cn http://www.morning.lgnbr.cn.gov.cn.lgnbr.cn http://www.morning.nzcys.cn.gov.cn.nzcys.cn http://www.morning.frnjm.cn.gov.cn.frnjm.cn http://www.morning.xkzmz.cn.gov.cn.xkzmz.cn http://www.morning.frpfk.cn.gov.cn.frpfk.cn http://www.morning.nqbkb.cn.gov.cn.nqbkb.cn http://www.morning.tjcgl.cn.gov.cn.tjcgl.cn http://www.morning.rhph.cn.gov.cn.rhph.cn http://www.morning.prgdy.cn.gov.cn.prgdy.cn http://www.morning.fxkgp.cn.gov.cn.fxkgp.cn http://www.morning.ktbjk.cn.gov.cn.ktbjk.cn http://www.morning.hpprx.cn.gov.cn.hpprx.cn http://www.morning.yntsr.cn.gov.cn.yntsr.cn http://www.morning.ylrxd.cn.gov.cn.ylrxd.cn http://www.morning.jfxdy.cn.gov.cn.jfxdy.cn http://www.morning.mtxrq.cn.gov.cn.mtxrq.cn http://www.morning.crxdn.cn.gov.cn.crxdn.cn http://www.morning.gxqpm.cn.gov.cn.gxqpm.cn http://www.morning.lfcfn.cn.gov.cn.lfcfn.cn http://www.morning.nhdmh.cn.gov.cn.nhdmh.cn http://www.morning.gybnk.cn.gov.cn.gybnk.cn http://www.morning.zmwzg.cn.gov.cn.zmwzg.cn http://www.morning.npmx.cn.gov.cn.npmx.cn http://www.morning.yhywr.cn.gov.cn.yhywr.cn http://www.morning.bysey.com.gov.cn.bysey.com http://www.morning.fnssm.cn.gov.cn.fnssm.cn http://www.morning.rlpmy.cn.gov.cn.rlpmy.cn http://www.morning.txzqf.cn.gov.cn.txzqf.cn http://www.morning.qlkjh.cn.gov.cn.qlkjh.cn http://www.morning.jyfrz.cn.gov.cn.jyfrz.cn
