【K8S】基于Kubernetes部署Kafka集群

主要参考了https://stackoverflow.com/questions/44651219/kafka-deployment-on-minikube和https://github.com/ramhiser/kafka-kubernetes两个项目，但是这两个项目都是单节点的Kafka，我这里尝试将单节点Kafka扩展为多节点的Kafka集群。

一、单节点Kafka

要搭建Kafka集群，还是要从单节点开始。

1.创建Zookeeper服务zookeeper-svc.yaml和zookeeper-deployment.yaml，用kubectl create -f创建：

apiVersion: v1
kind: Service
metadata:labels:app: zookeeper-servicename: zookeeper-service
spec:ports:- name: zookeeper-portport: 2181targetPort: 2181selector:app: zookeeper

apiVersion: extensions/v1beta1
kind: Deployment
metadata:labels:app: zookeepername: zookeeper
spec:replicas: 1template:metadata:labels:app: zookeeperspec:containers:- image: wurstmeister/zookeeperimagePullPolicy: IfNotPresentname: zookeeperports:- containerPort: 2181

2.等pod跑起来，service的endpoint配置成功后，就可以继续创建kafka的kafka-svc.yaml和kafka-deployment.yaml了：

apiVersion: v1
kind: Service
metadata:name: kafka-servicelabels:app: kafka
spec:type: NodePortports:- port: 9092name: kafka-porttargetPort: 9092nodePort: 30092protocol: TCPselector:app: kafka

kind: Deployment
apiVersion: extensions/v1beta1
metadata:name: kafka-deployment
spec:replicas: 1selector:matchLabels:name: kafkatemplate:metadata:labels:name: kafkaapp: kafkaspec:containers:- name: kafkaimage: wurstmeister/kafkaimagePullPolicy: IfNotPresentports:- containerPort: 9092env:- name: KAFKA_ADVERTISED_PORTvalue: "9092"- name: KAFKA_ADVERTISED_HOST_NAMEvalue: "[kafka的service的clusterIP]"- name: KAFKA_ZOOKEEPER_CONNECTvalue: [zookeeper的service的clusterIP]:2181- name: KAFKA_BROKER_IDvalue: "1"

clusterIP通过kubectl get svc进行查看。KAFKA_ZOOKEEPER_CONNECT的值也可以改为zookeeper-service:2181。

3.创建后，需要对服务进行测试。参考了https://blog.csdn.net/boling_cavalry/article/details/78309050的方法。

在此之前，针对虚拟化的Kafka，需要先执行下面的命令以进入容器：

kubectl exec -it [Kafka的pod名称] /bin/bash

进入容器后，Kafka的命令存储在opt/kafka/bin目录下，用cd命令进入：

cd opt/kafka/bin

后面的操作就跟上面的博客中写的类似了。针对单节点Kafka，需要将同一个节点作为生产者和消费者。执行命令如下：

kafka-console-producer.sh --broker-list [kafka的service的clusterIP]:9092 --topic test

运行正常的话，下方会出现>标记以提示输入消息。这样这个终端就成为了生产者。

另外打开一个linux终端，执行相同的命令进入容器。这次将这个终端作为消费者。注意，上面的博客中写的创建消费者的方法在新版的Kafka中已经改变，需要执行下面的命令：

kafka-console-consumer.sh --bootstrap-server [kafka的service的clusterIP]:9092 --topic test --from-beginning

之后，在生产者输入信息，查看消费者是否能够接收到。如果接收到，说明运行成功。

最后，还可以执行下面的命令以测试列出所有的消息主题：

kafka-topics.sh --list --zookeeper [zookeeper的service的clusterIP]:2181

注意，有时需要用Kafka的端口，有时需要用Zookeeper的端口，应注意区分。

二、多节点Kafka集群

单节点服务运行成功后，就可以尝试增加Kafka的节点以建立集群。我的Kubernetes集群包含3个节点，所以我搭建的Kafka集群也包含3个节点，分别运行在三台机器上。

我这里采用了3个Deployment来运行Kafka和Zookeeper，其实更优雅的方式是使用StatefulSet。Kubernetes的官方文档上有使用StatefulSet搭建Zookeeper集群的范例。

但是使用StatefulSet搭建Zookeeper和Kafka时，Zookeeper的myid和Kafka的brokerID就不能预先设置了，因此需要在镜像构建过程中加入相关的操作，而Docker Hub中的绝大多数镜像都不包含这一逻辑。而Deployment虽然不够优雅，但是可以对各节点预先配置，运行起来相对简单，可以说各有所长。

1.搭建Zookeeper集群

创建zookeeper的yaml文件zookeeper-svc2.yaml和zookeeper-deployment2.yaml如下：

apiVersion: v1
kind: Service
metadata:name: zoo1labels:app: zookeeper-1
spec:ports:- name: clientport: 2181protocol: TCP- name: followerport: 2888protocol: TCP- name: leaderport: 3888protocol: TCPselector:app: zookeeper-1
---
apiVersion: v1
kind: Service
metadata:name: zoo2labels:app: zookeeper-2
spec:ports:- name: clientport: 2181protocol: TCP- name: followerport: 2888protocol: TCP- name: leaderport: 3888protocol: TCPselector:app: zookeeper-2
---
apiVersion: v1
kind: Service
metadata:name: zoo3labels:app: zookeeper-3
spec:ports:- name: clientport: 2181protocol: TCP- name: followerport: 2888protocol: TCP- name: leaderport: 3888protocol: TCPselector:app: zookeeper-3

kind: Deployment
apiVersion: extensions/v1beta1
metadata:name: zookeeper-deployment-1
spec:replicas: 1selector:matchLabels:app: zookeeper-1name: zookeeper-1template:metadata:labels:app: zookeeper-1name: zookeeper-1spec:containers:- name: zoo1image: digitalwonderland/zookeeperimagePullPolicy: IfNotPresentports:- containerPort: 2181env:- name: ZOOKEEPER_IDvalue: "1"- name: ZOOKEEPER_SERVER_1value: zoo1- name: ZOOKEEPER_SERVER_2value: zoo2- name: ZOOKEEPER_SERVER_3value: zoo3
---
kind: Deployment
apiVersion: extensions/v1beta1
metadata:name: zookeeper-deployment-2
spec:replicas: 1selector:matchLabels:app: zookeeper-2name: zookeeper-2template:metadata:labels:app: zookeeper-2name: zookeeper-2spec:containers:- name: zoo2image: digitalwonderland/zookeeperimagePullPolicy: IfNotPresentports:- containerPort: 2181env:- name: ZOOKEEPER_IDvalue: "2"- name: ZOOKEEPER_SERVER_1value: zoo1- name: ZOOKEEPER_SERVER_2value: zoo2- name: ZOOKEEPER_SERVER_3value: zoo3
---
kind: Deployment
apiVersion: extensions/v1beta1
metadata:name: zookeeper-deployment-3
spec:replicas: 1selector:matchLabels:app: zookeeper-3name: zookeeper-3template:metadata:labels:app: zookeeper-3name: zookeeper-3spec:containers:- name: zoo3image: digitalwonderland/zookeeperimagePullPolicy: IfNotPresentports:- containerPort: 2181env:- name: ZOOKEEPER_IDvalue: "3"- name: ZOOKEEPER_SERVER_1value: zoo1- name: ZOOKEEPER_SERVER_2value: zoo2- name: ZOOKEEPER_SERVER_3value: zoo3

这里创建了3个deployment和3个service，一一对应。这样，三个实例都可以对外提供服务。

创建完成后，需要用kubectl logs查看一下三个Zookeeper的pod的日志，确保没有错误发生，并且在3个节点的日志中，有类似下面的语句，则表明Zookeeper集群已顺利搭建成功。

2020-06-16 00:00:05,904 [myid:2] - INFO [QuorumPeer[myid=2]/0:0:0:0:0:0:0:0:2181:Leader@358] - LEADING - <br>LEADER ELECTION TOOK - 2613

2.搭建Kafka集群

同样创建3个deployment和3个service，编写kafka-svc2.yaml和kafka-deployment2.yaml如下：

apiVersion: v1
kind: Service
metadata:name: kafka-service-1labels:app: kafka-service-1
spec:type: NodePortports:- port: 9092name: kafka-service-1targetPort: 9092nodePort: 30901protocol: TCPselector:app: kafka-service-1
---
apiVersion: v1
kind: Service
metadata:name: kafka-service-2labels:app: kafka-service-2
spec:type: NodePortports:- port: 9092name: kafka-service-2targetPort: 9092nodePort: 30902protocol: TCPselector:app: kafka-service-2
---
apiVersion: v1
kind: Service
metadata:name: kafka-service-3labels:app: kafka-service-3
spec:type: NodePortports:- port: 9092name: kafka-service-3targetPort: 9092nodePort: 30903protocol: TCPselector:app: kafka-service-3

kind: Deployment
apiVersion: extensions/v1beta1
metadata:name: kafka-deployment-1
spec:replicas: 1selector:matchLabels:name: kafka-service-1template:metadata:labels:name: kafka-service-1app: kafka-service-1spec:containers:- name: kafka-1image: wurstmeister/kafkaimagePullPolicy: IfNotPresentports:- containerPort: 9092env:- name: KAFKA_ADVERTISED_PORTvalue: "9092"- name: KAFKA_ADVERTISED_HOST_NAMEvalue: [kafka-service1的clusterIP]- name: KAFKA_ZOOKEEPER_CONNECTvalue: zoo1:2181,zoo2:2181,zoo3:2181- name: KAFKA_BROKER_IDvalue: "1"- name: KAFKA_CREATE_TOPICSvalue: mytopic:2:1
---
kind: Deployment
apiVersion: extensions/v1beta1
metadata:name: kafka-deployment-2
spec:replicas: 1selector:selector:matchLabels:name: kafka-service-2template:metadata:labels:name: kafka-service-2app: kafka-service-2spec:containers:- name: kafka-2image: wurstmeister/kafkaimagePullPolicy: IfNotPresentports:- containerPort: 9092env:- name: KAFKA_ADVERTISED_PORTvalue: "9092"- name: KAFKA_ADVERTISED_HOST_NAMEvalue: [kafka-service2的clusterIP]- name: KAFKA_ZOOKEEPER_CONNECTvalue: zoo1:2181,zoo2:2181,zoo3:2181- name: KAFKA_BROKER_IDvalue: "2"
---
kind: Deployment
apiVersion: extensions/v1beta1
metadata:name: kafka-deployment-3
spec:replicas: 1selector:selector:matchLabels:name: kafka-service-3template:metadata:labels:name: kafka-service-3app: kafka-service-3spec:containers:- name: kafka-3image: wurstmeister/kafkaimagePullPolicy: IfNotPresentports:- containerPort: 9092env:- name: KAFKA_ADVERTISED_PORTvalue: "9092"- name: KAFKA_ADVERTISED_HOST_NAMEvalue: [kafka-service3的clusterIP]- name: KAFKA_ZOOKEEPER_CONNECTvalue: zoo1:2181,zoo2:2181,zoo3:2181- name: KAFKA_BROKER_IDvalue: "3"

在deployment1中执行了创建一个新topic的操作。

3.测试

测试方法基本同单集群的情况，这里就不赘述了。不同的是，这次可以将不同的节点作为生产者和消费者。

至此，Kubernetes的Kafka集群搭建就大功告成了！