go语言kafka入门

消息队列:一种基于异步通信的解耦机制,用于在应用程序或系统组件之间传递消息和数据

消息队列相关概念:

生产者(Producer):生成并发送消息到消息队列中的应用程序或系统组件。

消费者(Consumer):从消息队列中接收和处理消息的应用程序或系统组件。

主题(Topic):消息队列中用于分类和分组消息的逻辑概念,生产者将消息发送到指定的主题,而消费者可以订阅特定的主题以接收相应的消息。

队列(Queue):消息队列中存储消息的容器,遵循先进先出(FIFO)的原则。

发布-订阅模式(Publish-Subscribe Pattern):一种消息分发模式,生产者将消息发送到一个或多个主题,而消费者通过订阅感兴趣的主题来接收相应的消息。

点对点模式(Point-to-Point Pattern):一种消息传递模式,生产者将消息发送到特定的队列中,而消费者从队列中接收并处理消息。

消息序列化(Message Serialization):将消息从应用程序的数据结构转换为可以在消息队列中传输和存储的格式,通常使用如JSON、XML 或二进制等格式。

消息持久化(Message Persistence):将消息保存到持久化存储中,以确保即使在消息队列或应用程序重启之后也不会丢失

在这里插入图片描述

安装kafka-go

go get github.com/segmentio/kafka-go

简单示例

package mainimport ("context""fmt""log""sync""time""github.com/segmentio/kafka-go"
)// writeByConn 基于Conn发送消息
func writeByConn(wg *sync.WaitGroup) {defer wg.Done()topic := "my-topic"partition := 0// 连接至Kafka集群的Leader节点conn, err := kafka.DialLeader(context.Background(), "tcp", "localhost: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)}
}// readByConn 连接至kafka后接收消息
func readByConn(wg *sync.WaitGroup) {defer wg.Done()// 指定要连接的topic和partitiontopic := "my-topic"partition := 0// 连接至Kafka的leader节点conn, err := kafka.DialLeader(context.Background(), "tcp", "localhost:9092", topic, partition)if err != nil {log.Fatal("failed to dial leader:", err)}// 设置读取超时时间conn.SetReadDeadline(time.Now().Add(10 * time.Second))// 读取一批消息,得到的batch是一系列消息的迭代器batch := conn.ReadBatch(10e3, 1e6) // fetch 10KB min, 1MB max// 遍历读取消息b := make([]byte, 10e3) // 10KB max per messagefor {n, err := batch.Read(b)if err != nil {break}fmt.Println(string(b[:n]))}// 关闭batchif 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)}
}func main() {var wg sync.WaitGroupwg.Add(2)go writeByConn(&wg)go readByConn(&wg)wg.Wait()
}

运行结果

one!
two!
three!
one!
two!
three!
one!
two!
three!
one!
two!
three!
one!
two!
three!
2023/08/25 10:14:10 failed to close batch:[7] Request Timed Out: the request exceeded the user-specified time limit in the request

可以看到,利用两个goroutine,成功在一个文件里实现了生产者写入消息,消费者消费消息并打印出来

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.rhkb.cn/news/115046.html

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

微服务事务管理(Dubbo)

Seata 是什么 Seata 是一款开源的分布式事务解决方案,致力于提供高性能和简单易用的分布式事务服务。Seata 将为用户提供了 AT、TCC、SAGA 和 XA 事务模式,为用户打造一站式的分布式解决方案。 一、示例架构说明 可在此查看本示例完整代码地址&#x…

【GO】LGTM_Grafana_Tempo(1)_架构

最近在尝试用 LGTM 来实现 Go 微服务的可观测性,就顺便整理一下文档。 Tempo 会分为 4 篇文章: Tempo 的架构官网测试实操跑通gin 框架发送 trace 数据到 tempogo-zero 微服务框架使用发送数据到 tempo 第一篇是关于,tempo 的架构&#xff…

0828|C++day6 菱形继承+虚继承+多态+抽象类+模板

一、思维导图 二、今日知识回顾 1&#xff09;多态 父类的指针或者引用&#xff0c;指向或初始化子类的对象&#xff0c;调用子类对父类重写的函数&#xff0c;进而展开子类的功能。 #include <iostream> using namespace std;class Zhou { private:string name;int age…

eclipse/STS(Spring Tool Suite)安装CDT环境(C/C++)

在线安装 help -> eclipse marketplace 可以发现&#xff0c;我所使用eclipse给我推荐安装的CDT是10.5版本 离线安装 下载离线安装包 下载地址&#xff1a;https://github.com/eclipse-cdt/cdt/blob/main/Downloads.md 可以看到利息安装包主要有如下四大类&#xff0c;…

【Kali Linux高级渗透测试】深入剖析Kali Linux:高级渗透测试技术与实践

&#x1f4d5;作者简介&#xff1a;热爱跑步的恒川&#xff0c;致力于C/C、Java、Python等多编程语言&#xff0c;热爱跑步&#xff0c;喜爱音乐的一位博主。 &#x1f4d7;本文收录于恒川的日常汇报系列&#xff0c;大家有兴趣的可以看一看 &#x1f4d8;相关专栏C语言初阶、C…

(vue)Vue项目中使用jsPDF和html2canvas生成PDF

(vue)Vue项目中使用jsPDF和html2canvas生成PDF 效果&#xff1a; 安装与使用 1.&#xff1a;安装jsPDF和html2canvas npm install jspdf html2canvas2.在需要生成PDF文档的组件中引入jsPDF和html2canvas <template><div><el-button type"primary"…

Django(7)-项目实战-发布会签到管理系统

本文使用django实现一个简单的发布会签到管理系统 登录功能 模板页面 sign/templates/index.html <!DOCTYPE html> <html> <head><title>Login Page</title> </head> <body><h1>发布会管理</h1><form action=&qu…

视频融合平台EasyCVR视频汇聚平台关于小区高空坠物安全实施应用方案设计

近年来&#xff0c;随着我国城市化建设的推进&#xff0c;高楼大厦越来越多&#xff0c;高空坠物导致的伤害也屡见不鲜&#xff0c;严重的影响到人们的生命安全。像在日常生活中一些不起眼的小东西如烟头、鸡蛋、果核、易拉罐&#xff0c;看似伤害不大&#xff0c;但只要降落的…

无涯教程-Android - DatePicker函数

Android Date Picker允许您在自定义用户界面中选择由日,月和年组成的日期。为此功能,android提供了DatePicker和DatePickerDialog组件。 在本教程中,我们将通过DatePickerDialog演示日期选择器的用法, DatePickerDialog是一个包含DatePicker的简单对话框。 为了显示DatePicker…

RTSP/Onvif视频服务器EasyNVR安防视频平台服务器频繁重启的问题解决方案

EasyNVR平台优秀的视频能力在于通过RTSP/ONVIF协议&#xff0c;将前端接入设备的音视频资源进行采集&#xff0c;并转码成适合全平台、全终端分发的视频流格式&#xff0c;包括RTSP、RTMP、FLV、HLS、WebRTC等格式。平台可拓展性强、部署轻快&#xff0c;在安防监控领域有着广泛…

决策树(Decision Tree)

决策树的定义: 分类决策树模型是一种描述对实例进行分类的树形结构。决策树由结点&#xff08;node&#xff09;和有向边&#xff08;directed edge&#xff09;组成。结点有两种类型: 内部结点&#xff08;internal node&#xff09;和叶结点&#xff08;leaf node&#xff0…

Linux的内存理解

建议 Mysql机器 尽量不要硬swap,如果是ssd磁盘还好。Free命令 free 命令显示系统内存的使用情况,包括物理内存、交换内存(swap)和内核缓冲区内存 输出简介: Mem 行(第二行)是内存的使用情况。Swap 行(第三行)是交换空间的使用情况。total 列显示系统总的可用物理内存和交换…

curl通过webdav操作alist

创建目录: url202320230828;curl -v -u "admin":"这里是密码" -X MKCOL "http://127.0.0.1:5244/dav/my189tianyi/${url2023}/" 上传文件: curl -v -u "admin":"这里是密码" -T /tmp/aa.json "http://127.0.0.1:52…

Android Activity 启动流程 二:setContentView

关于作者&#xff1a;CSDN内容合伙人、技术专家&#xff0c; 从零开始做日活千万级APP。 专注于分享各领域原创系列文章 &#xff0c;擅长java后端、移动开发、商业变现、人工智能等&#xff0c;希望大家多多支持。 目录 一、概览二、setContentView&#xff08;&#xff09;三…

springboot集成es 插入和查询的简单使用

第一步&#xff1a;引入依赖 <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-elasticsearch</artifactId><version>2.2.5.RELEASE</version></dependency>第二步&#xff1a;…

iTOP-RK3588开发板Android12 设置系统默认不休眠

修改文件&#xff1a; device/rockchip/rk3588/overlay/frameworks/base/packages/SettingsProvider/res/values/defaults. xml 文件&#xff0c;如下图所示&#xff1a; - <integer name"def_screen_off_timeout">60000</integer> <integer name&q…

什么是同源策略(same-origin policy)?它对AJAX有什么影响?

聚沙成塔每天进步一点点 ⭐ 专栏简介⭐ 同源策略&#xff08;Same-Origin Policy&#xff09;与 AJAX 影响⭐ 同源策略的限制⭐ AJAX 请求受同源策略影响⭐ 跨域资源共享&#xff08;CORS&#xff09;⭐ 写在最后 ⭐ 专栏简介 前端入门之旅&#xff1a;探索Web开发的奇妙世界 记…

小文智能GPT助手介绍

如何使用小文交互的GPT助手&#xff0c;让AI更加智能&#xff0c;适用更多场景&#xff1f; 在小文智能最新推出的4.0版本&#xff0c;有一个新功能&#xff0c;叫做GPT助手。GPT助手&#xff0c;顾名思义&#xff0c;即在小文智能的场景中&#xff0c;接入ChatGPT&#xff0c…

C++(17):异常处理

异常处理机制允许程序中独立开发的部分能够在运行时就出现的问题进行通信并做出相应的处理。 异常使得能够将问题的检测与解决过程分离开来&#xff1a;程序的一部分负责检测问题的出现&#xff0c;然后解决该问题的任务传递给程序的另一部分。检测环节无须知道问题处理模块的…

癌症预测新利器:弹性逻辑回归让健康更可控!

一、引言 癌症是全球范围内健康领域的重大挑战&#xff0c;早期预测和诊断对于提高治疗效果和生存率至关重要。在过去的几十年里&#xff0c;随着医学和数据科学的快速发展&#xff0c;基于机器学习和统计方法的癌症风险预测成为研究的热点。其中&#xff0c;弹性逻辑回归作为一…