Docker部署kafka|Go操作实践

本文涉及的产品
服务治理 MSE Sentinel/OpenSergo,Agent数量 不受限
简介: 本文讲解了使用docker-compose部署单节点kafka的流程,并且在文章的后半部分给出了使用Go语言操作kafka进行生产消费的代码案例。

前言

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

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

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

部署kafka

docker前置知识

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

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

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

docker-compose

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

version: '3'
services:
    zookeeper: 
        image: wurstmeister/zookeeper:3.4.6 
        volumes: 
            - ./zookeeper_data:/opt/zookeeper-3.4.6/data 
        container_name: zookeeper 
        ports: 
            - "10002:2181" 
            - "10003:2182" 
        restart: always
​
    kafka: 
        image: wurstmeister/kafka 
        container_name: kafka_01 
        depends_on: 
            - zookeeper 
        ports: 
            - "10004:9092" 
        volumes: 
            - ./kafka_log:/kafka 
        environment: 
            - KAFKA_BROKER_NO=0 
            - KAFKA_BROKER_ID=0 
            - KAFKA_LISTENERS=PLAINTEXT://kafka_01:9092                     # kafka tcp 侦听的ip
            - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://服务器ip:10004        # kafka broker侦听的ip
            - KAFKA_INTER_BROKER_LISTENER_NAME=PLAINTEXT 
            - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 
            - KAFKA_HEAP_OPTS=-Xmx512M -Xms16M
        restart: always
    # kafka集群管理面板
    kafka_manager: 
        image: sheepkiller/kafka-manager 
        ports: 
            - "10005:9000" 
        environment: 
            - ZK_HOSTS=zookeeper:2181 
        depends_on: 
            - zookeeper 
            - kafka 
        restart: always

后台运行

docker-compose up -d

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

image-20220804102532075

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

image-20220804101807534

docker exec -it kafka容器id /bin/bash
# 进入kafka目录
cd /opt/kafka_2.13-2.8.1/

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

bin/kafka-topics.sh --create --zookeeper 服务器ip:2181 --replication-factor 1 -partitions 1 --topic test

业务编写

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

go get github.com/segmentio/kafka-go

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

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

producer

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

func main() {
   config := sarama.NewConfig()
   config.Producer.RequiredAcks = sarama.WaitForAll          // 发送完数据需要leader和follow都确认
   config.Producer.Partitioner = sarama.NewRandomPartitioner // 新选出一个partition
   config.Producer.Return.Successes = true                   // 成功交付的消息将在success channel返回

   // 构造一个消息
   msg := &sarama.ProducerMessage{}
   msg.Topic = "test"
   // 连接kafka
   client, err := sarama.NewSyncProducer([]string{"82.156.171.8:10004"}, config)
   if err != nil {
      fmt.Println("producer closed, err:", err)
      return
   }
   defer client.Close()
   // 发送消息

   for {
      time.Sleep(time.Second * 1)
      msg.Value = sarama.StringEncoder("this is a test log")
      pid, offset, err := client.SendMessage(msg)
      if err != nil {
         fmt.Println("send msg failed, err:", err)
         return
      }
      fmt.Printf("pid:%v offset:%v\n", pid, offset)
   }

}

consumer

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

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

func main() {
   consumer, err := sarama.NewConsumer([]string{"82.156.171.8:10004"}, nil)
   if err != nil {
      fmt.Printf("fail to start consumer, err:%v\n", err)
      return
   }
   partitionList, err := consumer.Partitions("test") // 根据topic取到所有的分区
   if err != nil {
      fmt.Printf("fail to get list of partition:err%v\n", err)
      return
   }
   fmt.Println("list = ", partitionList, len(partitionList))
   var wg sync.WaitGroup
   for partition := range partitionList { // 遍历所有的分区
      wg.Add(1)
      // 针对每个分区创建一个对应的分区消费者
      pc, err := consumer.ConsumePartition("test", int32(partition), sarama.OffsetNewest)
      if err != nil {
         fmt.Printf("failed to start consumer for partition %d,err:%v\n", partition, err)
         return
      }
      defer pc.AsyncClose()
      go func(sarama.PartitionConsumer, *sync.WaitGroup) {
         for msg := range pc.Messages() {
            //fmt.Println("打印信息")
            fmt.Println("Partition:%d Offset:%d Key:%v Value:%v", msg.Partition, msg.Offset, msg.Key, string(msg.Value))
         }
         wg.Done()
      }(pc, &wg)
   }
   wg.Wait()
}

生产&消费

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

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

image-20220804120429711

小结

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

相关文章
|
2天前
|
Java Linux Maven
Linux系统Docker部署Nexus Maven并实现远程访问本地管理界面
Linux系统Docker部署Nexus Maven并实现远程访问本地管理界面
|
2天前
|
关系型数据库 MySQL 数据库
如何使用Docker部署开源CMF Drupal并结合cpolar内网穿透远程访问
如何使用Docker部署开源CMF Drupal并结合cpolar内网穿透远程访问
|
2天前
|
存储 Linux 数据安全/隐私保护
如何在本地Docker中部署MinIO服务并实现远程访问管理界面
如何在本地Docker中部署MinIO服务并实现远程访问管理界面
|
1天前
|
JavaScript 测试技术 数据安全/隐私保护
【Docker项目实战】使用Docker部署Mikochi文件管理工具
【2月更文挑战第12天】使用Docker部署Mikochi文件管理工具
21 5
|
2天前
|
存储 Linux 测试技术
【Docker项目实战】使用Docker部署Raneto知识库平台
【2月更文挑战第11天】使用Docker部署Raneto知识库平台
33 1
【Docker项目实战】使用Docker部署Raneto知识库平台
|
2天前
|
数据挖掘 Apache 数据安全/隐私保护
使用Docker部署Apache Superset并实现公网远程访问
使用Docker部署Apache Superset并实现公网远程访问
|
3天前
|
NoSQL 应用服务中间件 nginx
【Docker】3、Docker 基本操作【容器操作】
【Docker】3、Docker 基本操作【容器操作】
16 0
|
3天前
|
NoSQL 应用服务中间件 nginx
【Docker】2、Docker 基本操作【镜像操作】
【Docker】2、Docker 基本操作【镜像操作】
18 0
|
3天前
|
资源调度 JavaScript 前端开发
如何使用 Docker 来部署 Vue 项目?
【2月更文挑战第8天】
19 2
|
4天前
|
搜索推荐 测试技术 开发者
【Docker项目实战】使用Docker部署envlinks极简个人导航页
【2月更文挑战第9天】使用Docker部署envlinks极简个人导航页
32 1

相关产品

  • 云迁移中心