Docker部署kafka|Go操作实践

2022-08-18 14:31:45 浏览数 (1)

前言

写作本文的背景是由于字节的暑期青训营中,某个项目要求编写一个简易的流处理引擎(flink),开发语言不限,推荐Java,本着好奇心的驱使,我打算使用Go语言进行部分尝试。

既然是流处理引擎,那么首先需要有流式的数据源,一般而言,flink会配合从kafka中获取数据流,先不考虑后续编写引擎的部分,本文将着重于kafka的部署,并且后半段将给出使用Go语言编写kafka生产者消费者

如果你只是希望完成kafka的部署,而不想局限于Go语言,只需要着重阅读文章的前半部分,后文的Go语言操作部分可以给你提供一些思路,你只需要找寻适合语言如Javakafka client库去完成生产者消费者的编写即可。

部署kafka

docker前置知识

下文的实践需要你拥有基本的docker操作能力,如果未曾掌握docker知识点,推荐阅读这两篇文章:

docker | jenkins 实现自动化部署项目,后端躺着把运维的钱挣了!(上)

https://juejin.cn/post/7064024576497303565

docker | jenkins 自动化CI/CD,后端躺着把运维的钱挣了!(下)

https://juejin.cn/post/7064389514470359053

docker-compose

编写docker-compose.yml,通过docker容器部署单节点kafka

后台运行

docker ps命令查看容器是否启动成功

通过上述docker-compose.yml部署会运行三个容器,选择进入kafka容器

在容器内创建topictopic是kafka中数据管理的基本单位,或者说集合,每一个topic可以管理多个partition,编码操作时:你可以往对应kafka服务器ip port topic partition去发送和读取数据。

业务编写

Go语言中连接kafka使用第三方库: github.com/Shopify/sarama

sarama库的简易操作可以参照文档(消费者的编写文档中有坑):文档地址

如下使用kafka client库进行编码所涉及的API操作比较简单,流程上或许不够规范,请酌情参考。

producer

文档中生产者只发送了一条数据后就会关闭,这里我改成了每秒钟发送一次。

consumer

文档中消费者虽然开启了Go协程(类比于Java的线程)去读取kafka的数据,但是由于主程序执行顺序执行完毕后,子协程也会终止,导致子协程还没有读取成功/打印数据,整个程序就已经关闭运行了。

因此我做了一些改动,在子协程退出之前,保持主程序不会退出(使用Go语言的WaitGroup),如果简单粗暴在main函数末尾设置一个很长的程序sleep时间,也是可以实现打印输出的。

生产&消费

确保kafka容器正常运行,kafka服务器防火墙端口正常开放,运行消费者程序,运行生产者程序。这个生产者每秒向kafka发送一条测试数据:this is a test log,你也可以添加上程序运行时间进行测试。

事实上被客户端消费后的数据并没有马上从kafka删除,这里不多做介绍,各位自行了解~

小结

本文讲解了使用docker-compose部署单节点kafka的流程,后续通过修改docker-compose.yml的内容也可以实现kafka集群的部署,并且,在较新版本的kafka中,集群的部署可以脱离zookeeper,但是经过了解,由于功能并不完善,这里还是选择了基于zookeeper的部署。

0 人点赞