前言

最近在工作上调研k8s部署kafka集群的方案，顺带调研部署zookeeper方案。以下配置和部署方案都是亲测可用，有问题可以评论或私我解决。

kafka需要依赖zookeeper

kafka的生产者与消费者需要在zookeeper中注册，不然消费者怎么知道生产者是否存活之类的哈哈。废话不多说，直接上干货！

本文用的是statefulset和动态存储部署zookeeper和kafka集群。

部署zookeeper

apiVersion:v1kind:Servicemetadata:name:zk-headlessnamespace:liuleilabels:app:zkspec:ports:- port:2888name:server- port:3888name:leader-electionclusterIP:None #指定无头服务，需要对外暴露自行创建一个serviceselector:app:zk---apiVersion:policy/v1beta1kind:PodDisruptionBudgetmetadata:name:zk-pdbnamespace:liuleispec:selector:matchLabels:app:zkminAvailable:2---apiVersion:apps/v1kind:StatefulSetmetadata:name:zk #指定的zookeeper名称会在kafka.yaml里面用来配置和kafka连接，会创建pod为zk-0,zk-1,zk-2......这里只创建了三个namespace:liuleispec:selector:matchLabels:app:zkserviceName:zk-headlessreplicas:3updateStrategy:type:RollingUpdate # K8s 会将 StatefulSet 管理的 pod 分批次逐步替换掉podManagementPolicy:OrderedReady # 设置为Parallel这样Pod的创建就不必等待，而是会同时创建、同时删除template:metadata:labels:app:zkspec:containers:- name:k8s-zkimage:k8szk:1.0-3.4.10 #需要指定自己的镜像，可去阿里云上下载zookeeper镜像imagePullPolicy:Alwaysresources:requests:memory:"1Gi"cpu:"1000m"limits:memory:"1Gi"cpu:"1000m"ports:- containerPort:2181name:client- containerPort:2888name:server- containerPort:3888name:leader-electioncommand:- sh- -c- "start-zookeeper \--servers=5 \--data_dir=/var/lib/zookeeper/data \--data_log_dir=/var/lib/zookeeper/data/log \--conf_dir=/opt/zookeeper/conf \--client_port=2181 \--election_port=3888 \--server_port=2888 \--tick_time=2000 \--init_limit=10 \--sync_limit=5 \--heap=512M \--max_client_cnxns=60 \--snap_retain_count=3 \--purge_interval=12 \--max_session_timeout=40000 \--min_session_timeout=4000 \--log_level=INFO"- "zkGenConfig.sh && exec zkServer.sh start-foreground"readinessProbe:exec:command:- sh- -c- "zookeeper-ready 2181"initialDelaySeconds:10timeoutSeconds:5livenessProbe:exec:command:- sh- -c- "zookeeper-ready 2181"initialDelaySeconds:10timeoutSeconds:5imagePullPolicy:IfNotPresentvolumeMounts:- name:zookeeper-pvcmountPath:/var/lib/zookeepersecurityContext:runAsUser:1000fsGroup:1000volumeClaimTemplates:- metadata:name:zookeeper-pvclabels:type:statefulspec:accessModes:["ReadWriteMany"]storageClassName:storageclass-default #需要指定你自己的动态存储类名resources:requests:storage:1Gi

验证部署zookeeper

待zookeeper集群创建成功后验证zookeeper集群(注意修改自己的名称空间 -n后面参数)：

[root@master1 liulei]# for i in 0 1 2; do kubectl exec zk-$i -n liulei -- hostname; done

zk-0

zk-1

zk-2

[root@master1 liulei]# for i in 0 1 2; do echo "myid zk-$i";kubectl exec zk-$i -nliulei -- cat /var/lib/zookeeper/data/myid; done

myid zk-0

1

myid zk-1

2

myid zk-2

3

[root@master1 liulei]# for i in 0 1 2; do kubectl exec zk-$i -nliulei -- hostname -f; done

zk-0.zk-headless.liulei.svc.cluster.local

zk-1.zk-headless.liulei.svc.cluster.local

zk-2.zk-headless.liulei.svc.cluster.local

部署kafka

apiVersion:v1kind:Servicemetadata:name:kafka-svcnamespace:liuleilabels:app:kafkaspec:ports:- port:9093name:serverclusterIP:None #创建无头服务，如果需要对外暴露端口可自行创建serviceselector:app:kafka---apiVersion:policy/v1beta1kind:PodDisruptionBudgetmetadata:name:kafka-pdbnamespace:liuleispec:selector:matchLabels:app:kafkaminAvailable:2---apiVersion:apps/v1kind:StatefulSetmetadata:name:kafkanamespace:liuleispec:selector:matchLabels:app:kafkaserviceName:kafka-svcreplicas:3template:metadata:labels:app:kafkaspec:affinity:podAntiAffinity:requiredDuringSchedulingIgnoredDuringExecution:- labelSelector:matchExpressions:- key:"app"operator:Invalues:- kafkatopologyKey:"kubernetes.io/hostname"podAffinity:preferredDuringSchedulingIgnoredDuringExecution:- weight:1podAffinityTerm:labelSelector:matchExpressions:- key:"app"operator:Invalues:- zktopologyKey:"kubernetes.io/hostname"terminationGracePeriodSeconds:300containers:- name:k8skafkaimagePullPolicy:Alwaysimage:registry.cn-hangzhou.aliyuncs.com/ccgg/k8skafka:v1 #可自行去阿里云拉去对应镜像resources:requests:memory:"1Gi"cpu:"1000m"limits:memory:"1Gi"cpu:"1000m"ports:- containerPort:9093name:servercommand:- sh- -c- "exec kafka-server-start.sh /opt/kafka/config/server.properties --override broker.id=${HOSTNAME##*-} \--override listeners=PLAINTEXT://:9093 \--override zookeeper.connect=zk-0.zk-headless.liulei.svc.cluster.local:2181,zk-1.zk-headless.liulei.svc.cluster.local:2181,zk-2.zk-headless.liulei.svc.cluster.local:2181 \ #这里配置了与zookeeper进行连接，非常重要。格式为：pod名.zookeeper的service名.名称空间.svc.cluster.local:2181--override log.dir=/var/lib/kafka "env:- name:KAFKA_HEAP_OPTSvalue:"-Xmx512M -Xms512M"- name:KAFKA_OPTSvalue:"-Dlogging.level=INFO"volumeMounts:- name:kafka-datamountPath:/var/lib/kafkareadinessProbe:exec:command:- sh- -c- "/opt/kafka/bin/kafka-broker-api-versions.sh --bootstrap-server=localhost:9093"securityContext:runAsUser:1000fsGroup:1000volumeClaimTemplates:- metadata:name:kafka-dataspec:accessModes:["ReadWriteMany"]storageClassName:storageclass-default #需指定自己的动态存储类名resources:requests:storage:1Gi

部署成功后查看pod情况：

验证kafka是否可用：

1、进入kafka-0命令: kubectl exec -it kafka-0 -nliulei bash

进入容器目录：cd /opt/kafka/config

2、创建一个名为aaa的topc命令：kafka-topics.sh --create --topic aaa --zookeeper zk-0.zk-headless.liulei.svc.cluster.local:2181,zk-1.zk-headless.liulei.svc.cluster.local:2181,zk-2.zk-headless.liulei.svc.cluster.local:2181 --partitions 3 --replication-factor 2

结果为：

Created topic “aaa”.

3、进入topic为aaa的生产者消息中心：kafka-console-consumer.sh --topic aaa --bootstrap-server localhost:9093

4、复制新的会话，进入另一个容器kafka-1：kubectl exec -it kafka-1 -nliulei bash

进入消费者，输入命令：kafka-console-producer.sh --topic aaa --broker-list localhost:9093

输入:

hello

i lovle you

回车后，可在生产者消息中心看到消息

总结

以上步骤和内容都是我一个一个坑踩过来的，亲测可用，如果本文对你有帮助的话可以给我点个赞支持一下吗，比心❤。

欢迎访问我的博客，里面或许会有你感兴趣的文章哦leige24的博客_CSDN博客-K8S,Java,WSO2领域博主​blog.csdn.net