mkdir -p /usr/local/k8s/zookeeper/pv{1..3}
mkdir -p /usr/local/k8s/kafka/pv{1..3}

（2）编辑 /etc/exports 文件：

vi /etc/exports

在里面添加如下内容：

/usr/local/k8s/kafka/pv1 *(rw,sync,no_root_squash)
/usr/local/k8s/kafka/pv2 *(rw,sync,no_root_squash)
/usr/local/k8s/kafka/pv3 *(rw,sync,no_root_squash)
/usr/local/k8s/zookeeper/pv1 *(rw,sync,no_root_squash)
/usr/local/k8s/zookeeper/pv2 *(rw,sync,no_root_squash)
/usr/local/k8s/zookeeper/pv3 *(rw,sync,no_root_squash)

4）保存退出后执行如下命令重启服务：

如果执行 systemctl restart nfs 报“Failed to restart nfs.service: Unit nfs.service not found.”错误，可以尝试改用如下命令：

• sudo service nfs-server start

• systemctl restart rpcbind
  systemctl restart nfs
  systemctl enable nfs

（5）执行 exportfs -v 命令可以显示出所有的共享目录：

 （6）而其他的 Node 节点上需要执行如下命令安装 nfs-utils 客户端

yum -y install nfs-utils

（7）然后其他的 Node 节点上可执行如下命令（ip 为 Master 节点 IP）查看 Master 节点上共享的文件夹：

showmount -e  192.168.91.166（nfs服务端的IP）

1，创建 ZooKeeper PV

（1）首先创建一个 zookeeper-pv.yaml 文件，内容如下：

192.168.91.166

apiVersion: v1
kind: PersistentVolume
metadata:
name: k8s-pv-zk01
labels:
app: zk
spec:
storageClassName: anything
capacity:
storage: 1Gi
accessModes:
- ReadWriteOnce
nfs:
server: 192.168.91.166
path: "/usr/local/k8s/zookeeper/pv1"
persistentVolumeReclaimPolicy: Retain
---
apiVersion: v1
kind: PersistentVolume
metadata:
name: k8s-pv-zk02
labels:
app: zk
spec:
storageClassName: anything
capacity:
storage: 1Gi
accessModes:
- ReadWriteOnce
nfs:
server: 192.168.91.166
path: "/usr/local/k8s/zookeeper/pv2"
persistentVolumeReclaimPolicy: Retain
---
apiVersion: v1
kind: PersistentVolume
metadata:
name: k8s-pv-zk03
labels:
app: zk
spec:
storageClassName: anything
capacity:
storage: 1Gi
accessModes:
- ReadWriteOnce
nfs:
server: 192.168.91.166
path: "/usr/local/k8s/zookeeper/pv3"
persistentVolumeReclaimPolicy: Retain

（2）然后执行如下命令创建 PV：

kubectl apply -f zookeeper-pv.yaml

（3）执行如下命令可以查看是否创建成功：

2，创建 ZooKeeper 集群

 （1）搭建一个包含 3 个节点的 ZooKeeper 集群。创建 zookeeper.yaml 文件，内容如下

apiVersion: v1
kind: Service
metadata:
name: zk-hs
labels:
app: zk
spec:
selector:
app: zk
clusterIP: None
ports:
- name: server
port: 2888
- name: leader-election
port: 3888
---
apiVersion: v1
kind: Service
metadata:
name: zk-cs
labels:
app: zk
spec:
selector:
app: zk
type: NodePort
ports:
- name: client
port: 2181
nodePort: 31811
---
apiVersion: apps/v1
kind: StatefulSet
metadata:
name: zk
namespace: kafka
spec:
serviceName: "zk-hs"
replicas: 3
selector:
matchLabels:
app: zk
updateStrategy:
type: RollingUpdate
podManagementPolicy: Parallel
template:
metadata:
labels:
app: zk
spec:
containers:
- name: zk
imagePullPolicy: IfNotPresent
image: swr.cn-north-4.myhuaweicloud.com/ddn-k8s/docker.io/mirrorgooglecontainers/kubernetes-zookeeper:1.0-3.4.10
ports:
- containerPort: 2181
name: client
- containerPort: 2888
name: server
- containerPort: 3888
name: leader-election
command:
- sh
- -c
- "start-zookeeper \
--servers=3 \
--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=4G \
--max_client_cnxns=60 \
--snap_retain_count=3 \
--purge_interval=12 \
--max_session_timeout=40000 \
--min_session_timeout=4000 \
--log_level=INFO"
readinessProbe:
exec:
command:
- sh
- -c
- "zookeeper-ready 2181"
initialDelaySeconds: 10
timeoutSeconds: 5
livenessProbe:
exec:
command:
- sh
- -c
- "zookeeper-ready 2181"
initialDelaySeconds: 10
timeoutSeconds: 5
volumeMounts:
- name: datadir
mountPath: /var/lib/zookeeper
volumeClaimTemplates:
- metadata:
name: datadir
annotations:
volume.beta.kubernetes.io/storage-class: "anything"
spec:
accessModes: [ "ReadWriteOnce" ]
resources:
requests:
storage: 1Gi

（2）然后执行如下命令开始创建：

kubectl apply -f zookeeper.yaml

（3）执行如下命令可以查看是否创建成功：

kubectl get pods
kubectl get service

查看集群状态

# 观察pod启动
kubectl get pods -n kafka -w

# 查看自动生成PVC，确认Bound绑定PV
kubectl get pvc -n kafka

# 查看service
kubectl get svc -n kafka

# 验证zk集群连通性
kubectl exec -n kafka zk-0 -- zkCli.sh -server zk-0.zk-hs:2181 stat

 直接看 zookeeper 服务日志，看选举结果

kubectl logs -n kafka zk-0 | grep -E "LEADING|FOLLOWING"
kubectl logs -n kafka zk-1 | grep -E "LEADING|FOLLOWING"
kubectl logs -n kafka zk-2 | grep -E "LEADING|FOLLOWING"

另一种验证方式，进入 pod 交互式 shell

进入 pod 内部执行：

kubectl exec -it -n kafka zk-0 -- /bin/bash

echo stat | nc 127.0.0.1 2181

kubectl exec -it -n kafka zk-2 -- /bin/bash

echo stat | nc 127.0.0.1 2181

输出中会有 Mode: leader/follower

三、创建 Kafka 集群

（1）搭建一个包含 3 个节点的 Kafka 集群。创建一个 kafka.yaml 文件，内容如下：

apiVersion: v1
kind: Service
metadata:
name: kafka-service-1
labels:
app: kafka-1
spec:
type: NodePort
ports:
- port: 9092
name: kafka-service-1
targetPort: 9092
nodePort: 30901
protocol: TCP
selector:
app: kafka-1
---
apiVersion: v1
kind: Service
metadata:
name: kafka-service-2
labels:
app: kafka-2
spec:
type: NodePort
ports:
- port: 9092
name: kafka-service-2
targetPort: 9092
nodePort: 30902
protocol: TCP
selector:
app: kafka-2
---
apiVersion: v1
kind: Service
metadata:
name: kafka-service-3
labels:
app: kafka-3
spec:
type: NodePort
ports:
- port: 9092
name: kafka-service-3
targetPort: 9092
nodePort: 30903
protocol: TCP
selector:
app: kafka-3
---
apiVersion: apps/v1
kind: Deployment
metadata:
name: kafka-deployment-1
spec:
replicas: 1
selector:
matchLabels:
app: kafka-1
template:
metadata:
labels:
app: kafka-1
spec:
# ==========关键：禁用k8s自动注入service环境变量，防止wurstmeister脚本乱解析==========
enableServiceLinks: false
containers:
- name: kafka-1
image: swr.cn-north-4.myhuaweicloud.com/ddn-k8s/docker.io/wurstmeister/kafka:latest
imagePullPolicy: IfNotPresent
ports:
- containerPort: 9092
env:
- name: KAFKA_ZOOKEEPER_CONNECT
value: zk-0.zk-hs:2181,zk-1.zk-hs:2181,zk-2.zk-hs:2181
- name: KAFKA_BROKER_ID
value: "1"
- name: KAFKA_LISTENERS
value: PLAINTEXT://0.0.0.0:9092
- name: KAFKA_ADVERTISED_HOST_NAME
valueFrom:
fieldRef:
fieldPath: status.hostIP
- name: KAFKA_ADVERTISED_PORT
value: "30901"
volumeMounts:
- name: datadir
mountPath: /var/lib/kafka
volumes:
- name: datadir
nfs:
server: 192.168.91.166
path: "/usr/local/k8s/kafka/pv1"
---
apiVersion: apps/v1
kind: Deployment
metadata:
name: kafka-deployment-2
spec:
replicas: 1
selector:
matchLabels:
app: kafka-2
template:
metadata:
labels:
app: kafka-2
spec:
enableServiceLinks: false
containers:
- name: kafka-2
image: swr.cn-north-4.myhuaweicloud.com/ddn-k8s/docker.io/wurstmeister/kafka:latest
imagePullPolicy: IfNotPresent
ports:
- containerPort: 9092
env:
- name: KAFKA_ZOOKEEPER_CONNECT
value: zk-0.zk-hs:2181,zk-1.zk-hs:2181,zk-2.zk-hs:2181
- name: KAFKA_BROKER_ID
value: "2"
- name: KAFKA_LISTENERS
value: PLAINTEXT://0.0.0.0:9092
- name: KAFKA_ADVERTISED_HOST_NAME
valueFrom:
fieldRef:
fieldPath: status.hostIP
- name: KAFKA_ADVERTISED_PORT
value: "30902"
volumeMounts:
- name: datadir
mountPath: /var/lib/kafka
volumes:
- name: datadir
nfs:
server: 192.168.91.166
path: "/usr/local/k8s/kafka/pv2"
---
apiVersion: apps/v1
kind: Deployment
metadata:
name: kafka-deployment-3
spec:
replicas: 1
selector:
matchLabels:
app: kafka-3
template:
metadata:
labels:
app: kafka-3
spec:
enableServiceLinks: false
containers:
- name: kafka-3
image: swr.cn-north-4.myhuaweicloud.com/ddn-k8s/docker.io/wurstmeister/kafka:latest
imagePullPolicy: IfNotPresent
ports:
- containerPort: 9092
env:
- name: KAFKA_ZOOKEEPER_CONNECT
value: zk-0.zk-hs:2181,zk-1.zk-hs:2181,zk-2.zk-hs:2181
- name: KAFKA_BROKER_ID
value: "3"
- name: KAFKA_LISTENERS
value: PLAINTEXT://0.0.0.0:9092
- name: KAFKA_ADVERTISED_HOST_NAME
valueFrom:
fieldRef:
fieldPath: status.hostIP
- name: KAFKA_ADVERTISED_PORT
value: "30903"
volumeMounts:
- name: datadir
mountPath: /var/lib/kafka
volumes:
- name: datadir
nfs:
server: 192.168.91.166
path: "/usr/local/k8s/kafka/pv3"

（2）然后执行如下命令开始创建：

kubectl apply -f zookeeper-kafka-cluster.yaml

（3）执行如下命令可以查看是否创建成功：

kubectl get pods

kubectl get service

四、开始测试

1，K8s 集群内部测试

 （1）首先执行如下命令进入一个容器：

    kubectl exec -it kafka-deployment-1-6966544c7-2bwwb  /bin/bas

 （2）接着执行如下命令创建一个名为 test_topic 的 topic：

   kafka-topics.sh --create --topic test_topic --zookeeper zk-0.zk-hs:2181,zk-1.zk-hs:2181,zk-2.zk-hs:2181 --partitions 1 --replication-factor 1

 （3）创建后执行如下命令开启一个生产者，启动后可以直接在控制台中输入消息来发送，控制台中的每一行数据都会被视为一条消息来发送。

kafka-console-producer.sh --broker-list kafka-service-1:9092,kafka-service-2:9092,kafka-service-3:9092 --topic test_topic

（4）重新再打开一个终端连接服务器，然后进入容器后执行如下命令开启一个消费者：

kafka-console-consumer.sh --bootstrap-server kafka-service-1:9092,kafka-service-2:9092,kafka-service-3:9092 --topic test_topic

2，集群外出测试

    使用 Kafka 客户端工具（Offset Explorer）连接 Kafka 集群（可以通过 zookeeper 地址连接，也可以通过 kafka 地址连接），可以连接成功并能查看到数据。

更多的测试命令参考

二、Kafka生产者消费者实例（基于命令行）
1.创建一个itcasttopic的主题
代码如下（示例）：
kafka-topics.sh --create --topic itcasttopic --partitions 3 --replication-factor 2 -zookeeper zk-0.zk-hs:2181,zk-1.zk-hs:2181,zk-2.zk-hs:2181
　### 获取所有topic  kafka-topics.sh --list --zookeeper zk-0.zk-hs:2181,zk-1.zk-hs:2181,zk-2.zk-hs:2181
2.hadoop01当生产者
代码如下（示例）：
kafka-console-producer.sh --broker-list kafka-service-1:9092,kafka-service-2:9092,kafka-service-3:9092 --topic itcasttopic

3.hadoop02当消费者
代码如下（示例）：
kafka-console-consumer.sh --from-beginning --topic itcasttopic --bootstrap-server kafka-service-1:9092,kafka-service-2:9092,kafka-service-3:9092

3.–list查看所有主题
代码如下（示例）：
kafka-topics.sh --list --zookeeper zk-0.zk-hs:2181,zk-1.zk-hs:2181,zk-2.zk-hs:2181

4.删除主题
代码如下（示例）：
kafka-topics.sh --delete --zookeeper zk-0.zk-hs:2181,zk-1.zk-hs:2181,zk-2.zk-hs:2181 --topic itcasttopic

5.关闭kafka
代码如下（示例）：
bin/kafka-server-stop.sh config/server.properties