Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统，它可以处理消费者在网站中的所有动作流数据。 这种动作（网页浏览，搜索和其他用户的行动）是在现代网络上的许多社会功能的一个关键因素。 这些数据通常是由于吞吐量的要求而通过处理日志和日志聚合来解决。 对于像Hadoop一样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是通过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了通过集群来提供实时的消息。
体系结构图：
Send消息流程图：
Kafka多副本（Replica）机制：
如上图所示，Kafka 集群中有4个 broker，某个主题中有3个分区，且副本因子（即副本个数）也为3，如此每个分区便有1个 leader 副本和2个 follower 副本。生产者和消费者只与 leader 副本进行交互，而 follower 副本只负责消息的同步，很多时候 follower 副本中的消息相对 leader 副本而言会有一定的滞后。
ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务，是Google的Chubby一个开源的实现，是Hadoop和Hbase的重要组件。它是一个为分布式应用提供一致性服务的软件，提供的功能包括：配置维护、域名服务、分布式同步、组服务等。
原理：
高可以用架构图：
图中每一个Server代表一个安装Zookeeper服务的服务器。组成 ZooKeeper 服务的服务器都会在内存中维护当前的服务器状态，并且每台服务器之间都互相保持着通信。集群间通过 Zab 协议（Zookeeper Atomic Broadcast）来保持数据的一致性。
我们选择的是官方的chart地址：https://github.com/helm/charts/tree/master/incubator/kafka
1）编写自己的values.yaml文件（注意我的storageClass是已经做好了的nfs存储）
imageTag: "5.2.2"
resources:
limits:
cpu: 2
memory: 4Gi
requests:
cpu: 1
memory: 2Gi
kafkaHeapOptions: "-Xmx2G -Xms2G"
persistence:
enabled: true
storageClass: "managed-nfs-storage"
size: "40Gi"
zookeeper:
resources:
limits:
cpu: 1
memory: 2Gi
requests:
cpu: 100m
memory: 536Mi
persistence:
enabled: true
storageClass: "managed-nfs-storage"
size: "10Gi"
2）安装kafka
添加chart仓库：
$ helm repo add incubator http://storage.googleapis.com/kubernetes-charts-incubator
部署
$ helm install --name kafka -f values.yaml incubator/kafka
最后我们能看到：
1）根据提示创建一个测试客户端
apiVersion: v1
kind: Pod
metadata:
name: testclient
namespace: sscp-test
spec:
containers:
- name: kafka
image: solsson/kafka:0.11.0.0
command:
- sh
- -c
- "exec tail -f /dev/null"
Once you have the testclient pod above running, you can list all kafka
topics with:
kubectl -n sscp-test exec testclient -- kafka-topics --zookeeper kafka-test-zookeeper:2181 --list
To create a new topic:
kubectl -n sscp-test exec testclient -- kafka-topics --zookeeper kafka-test-zookeeper:2181 --topic test1 --create --partitions 1 --replication-factor 1
To listen for messages on a topic:
kubectl -n sscp-test exec -ti testclient -- for x in {1..1000}; do echo $x; sleep 2; done | kafka-console-producer --broker-list kafka-test-headless:9092 --topic test1
To stop the listener session above press: Ctrl+C
To start an interactive message producer session:
kubectl -n sscp-test exec -ti testclient -- kafka-console-producer --broker-list kafka-test-headless:9092 --topic test1
To create a message in the above session, simply type the message and press "enter"
To end the producer session try: Ctrl+C
注意：有三个kafka节点，消息要发三个副本才能保持其高可用！！！
1.Create a node by command below:
“kubectl exec -it testclient bash -n sscp-test”
“zookeeper-shell kafka-test-zookeeper-headless:2181 create /foo bar”
2. Check zookeeper status
Watch existing members:
$ kubectl run --attach bbox --image=busybox --restart=Never -- sh -c ‘while true; do for i in 0 1 2; do echo zk-${i} $(echo stats | nc kafka-zookeeper-${i}.kafka-zookeeper-headless:2181 | grep Mode); sleep 1; done; done‘
zk-2 Mode: follower
zk-0 Mode: follower
zk-1 Mode: leader
zk-2 Mode: follower
3.kill the leader by command below:
“Kubectl delete pod kafka-test-zookeeper-1”
4.Check the previously inserted key by command below:
““kubectl exec -it testclient bash -n sscp-test”
“zookeeper-shell kafka-test-zookeeper-headless:2181 get /foo”
原文：https://www.cnblogs.com/Dev0ps/p/11259324.html