一、概述

在k8s里面部署kafka、zookeeper这种有状态的服务，不能使用deployment和RC，k8s提供了一种专门用来部署这种有状态的服务的API--statefulset，有状态简单来说就是需要持久化数据，比如日志、数据库数据、服务状态等。

statefulset应用场景：

• 稳定的持久化存储，即Pod重新调度后还是能访问到相同的持久化数据，基于PVC来实现

• 稳定的网络标志，即Pod重新调度后其PodName和HostName不变，基于Headless Service（即没有Cluster IP的Service）来实现

• 有序部署，有序扩展，即Pod是有顺序的，在部署或者扩展的时候要依据定义的顺序依次依次进行（即从0到N-1，在下一个Pod运行之前所有之前的Pod必须都是Running和Ready状态），基于init containers来实现

• 有序收缩，有序删除（即从N-1到0）

statefulset组成：

• 用于定义网络标志（DNS domain）的Headless Service

• 用于创建PersistentVolumes的volumeClaimTemplates

• 定义具体应用的StatefulSet

StatefulSet中每个Pod的DNS格式为statefulSetName-{0..N-1}.serviceName.namespace.svc.cluster.local，其中 

• 0..N-1为Pod所在的序号，从0开始到N-1

• serviceName为Headless Service的名字

• statefulSetName为StatefulSet的名字

• namespace为服务所在的namespace，Headless Servic和StatefulSet必须在相同的namespace

• .cluster.local为Cluster Domain

二、部署

首先下载相关的部署文件：

git  clone  https://github.com/cuishuaigit/k8s-kafka.git

部署前需要提前准备nfs提供动态存储。

cd  k8s-kafka

kubectl create -f namesapce.yaml

kubectl create -f zk.yaml 
注意将storageClass的name换成自己的,默认是nfs-storage,除非存储名字一样。


kubectl create -f kafka.yaml

注意:  registry.cn-hangzhou.aliyuncs.com/syman-nj/kafka ,镜像源可以使用这个，自带的镜像源使用的是docker上面的，
会出现拉取镜像失败或是超慢。

详细信息参考：https://github.com/cuishuaigit/k8s-kafka

k8s-kafka

Requirements

k8s1.13.2+ cluster 正常运行，nfs服务用于提供持久化存储

kafka-zk version

kafka-2.2.0、scala-2.12、zk-3.4.10

Build image

sh run.sh

这里run.sh是直接上传到我的dockerhub，可以把地址改成自己的私有仓库地址

Deploymet

提前准备好nfs服务用于提供持久化存储(注意将storageClass的name换成自己的)，参考我的: https://github.com/cuishuaigit/k8s-monitor need deploy kafka and zookeeper to a special node,so you can use taint、toleration and label setting your node and statefulset

kubectl taint node [node-name] travis.io/schedule-only=kafka:NoSchedule
kubectl label node [node-name] travis.io/schedule-only=kafka
kubectl create -f namesapce.yaml
kubectl  create -f zk.yaml
kubectl  create -f kafka.yaml

Testing

部署在kafka这个namesapce里面。
测试zk：
kubectl exec -it zk-0 -- zkServer.sh status
kubectl exec -it zk-0 -- zkCli.sh create /hello world
kubectl delete -f zk.yaml 
kubectl apply -f zk.yaml
kubectl exec -it zk-0 -- zkCli.sh get /hello

测试kafka：
kubectl exec -it kafka-0 -- bash 

>kafka-topics.sh --create \
--topic test \
--zookeeper zk-0.zk-hs.kafka.svc.cluster.local:2181,zk-1.zk-hs.kafka.svc.cluster.local:2181,zk-2.zk-hs.kafka.svc.cluster.local:2181 \
--partitions 3 \
--replication-factor 2

>kafka-topics.sh --list --zookeeper zk-0.zk-hs.kafka.svc.cluster.local:2181,zk-1.zk-hs.kafka.svc.cluster.local:2181,zk-2.zk-hs.kafka.svc.cluster.local:2181

>kafka-console-consumer.sh --topic test --bootstrap-server localhost:9092

kubectl exec -it kafka-1 -- bash

>kafka-console-producer.sh --topic test --broker-list localhost:9092
随便输入一些东西：hello 

此时会在kafka-0的那边启动的kafka-console-consumer.sh会有相应的输出

线上使用的时候注意修改资源，如mem、cpu、磁盘空间大小。根据自己的需求进行修改，本次创建的是3节点的集群，如有其他需求，直接scale进行扩容，注意
不要超过nfs提供的空间大小，否则会启动失败，我使用的storageClass动态申请资源。

service FQDN

zk:
zk-0.zk-hs.kafka.svc.cluster.local
zk-1.zk-hs.kafka.svc.cluster.local
zk-2.zk-hs.kafka.svc.cluster.local 

zk-port:
2181

kafka:
kafka-0.kafka-svc.kafka.svc.cluster.local 
kafka-1.kafka-svc.kafka.svc.cluster.local
kafka-2.kafka-svc.kafka.svc.cluster.local 

kafka-port:
9092

Statefulset

参考：https://kubernetes.io/docs/tutorials/stateful-application/

我的 kafka 与zookeeper 6个容器，在一个zookeeper 建个主题，其它的都能收到的。

一个 消息生产者发送信息到一个kafka节点，其它的 节点，消费客户端都能拉取到 信息。