一、kafka各成员
kafka： 分布式消息系统，将消息直接存入磁盘，默认保存一周。broker： 组成kafka集群的节点，之间没有主从关系，依赖zookeeper来协调，broker负责满息的读写和存储，一个broker可以管理多个partition.producer： 消息的生产者，自己决定向哪个partition中去生产消息，两种机利：hash，轮循。consumer： 消息的消费者，consumer通过zookeeper去维护消费者偏移量。consumer有自己的消费者组，不同的组之间消费同一个topic数据，互不影响， 相同的组内的不同的consumer消费同一个topic，这个topic中相同的数据只能被消费一次。topic： 一类消息总称/一个消息队列。topic是由partition组成的，有多少？-> 创建指定。partition： 组成topic的单元，每个partition有副本，有多少？-> 创建topic时指定。每个partition只能由一个broker来管理，这个broker就是这个partition的leader。zookeeper： 协调kafka broker，存储原数据：consumer的offset+broker信息+topic信息+partition信息；
二、分析zookeeper元数据存储
1、创建topic
[root@spark1 bin]# pwd/usr/local/kafka/bin#创建topic[root@spark1 bin]# ./kafka-topics.sh --zookeeper spark1:2181,spark2,spark3 --create --topic t0425 --partitions 3 --replication-factor 3Created topic "t0425".[root@spark1 bin]# ./kafka-topics.sh --zookeeper spark1:2181,spark2,spark3 --list |grep t0425t0425
2、生产、消费
##生产者[root@spark1 bin]# ./kafka-console-producer.sh --topic t0425 --broker-list spark1:9092,spark2:9092,spark3:9092hello world123...##消费者[root@spark1 bin]# ./kafka-console-consumer.sh --zookeeper spark1:2181,spark2:2181,spark3:2181 --topic t0425hello world123...
3、进入zookeeper查看
看topic
##连接zookeeper[root@spark1 zk]# ./bin/zkCli.sh##查看[zk: localhost:2181(CONNECTED) 0] ls /[consumers, config, controller, admin, brokers, zookeeper, controller_epoch]#brokers下有topics[zk: localhost:2181(CONNECTED) 1] ls /brokers[topics, ids]#topics下有t0425，也就是刚才创建的topic[zk: localhost:2181(CONNECTED) 2] ls /brokers/topics[t0425, WordCount, TestTopic, WordCount1, first1]#t0425下有partitions[zk: localhost:2181(CONNECTED) 3] ls /brokers/topics/t0425[partitions][zk: localhost:2181(CONNECTED) 4] ls /brokers/topics/t0425/partitions[2, 1, 0][zk: localhost:2181(CONNECTED) 5] ls /brokers/topics/t0425/partitions/0[state][zk: localhost:2181(CONNECTED) 6] get /brokers/topics/t0425/partitions/0/state{ "controller_epoch":1,"leader":0,"version":1,"leader_epoch":0,"isr":[0,2,1]}cZxid = 0x2000000cectime = Fri Aug 23 10:06:02 CST 2019mZxid = 0x2000000cemtime = Fri Aug 23 10:06:02 CST 2019pZxid = 0x2000000cecversion = 0dataVersion = 0aclVersion = 0ephemeralOwner = 0x0dataLength = 76numChildren = 0
看consumer
[zk: localhost:2181(CONNECTED) 8] ls /consumers[console-consumer-72909, DefaultConsumerGroup][zk: localhost:2181(CONNECTED) 9] ls /consumers/console-consumer-72909[offsets, owners, ids][zk: localhost:2181(CONNECTED) 10] ls /consumers/console-consumer-72909/offsets [t0425]#可看到有0 1 2 [zk: localhost:2181(CONNECTED) 11] ls /consumers/console-consumer-72909/offsets/t0425[2, 1, 0][zk: localhost:2181(CONNECTED) 12] ls /consumers/console-consumer-72909/offsets/t0425/0[]#查看0[zk: localhost:2181(CONNECTED) 13] get /consumers/console-consumer-72909/offsets/t0425/012 #这个12就是刚才生产消费的消息条数，都在0里面,1和2里面没有cZxid = 0x2000000e7ctime = Fri Aug 23 10:14:09 CST 2019mZxid = 0x2000000e7mtime = Fri Aug 23 10:14:09 CST 2019pZxid = 0x2000000e7cversion = 0dataVersion = 0aclVersion = 0ephemeralOwner = 0x0dataLength = 2numChildren = 0##现在接着去生产几条消息，然后再看0 1 2[zk: localhost:2181(CONNECTED) 18] get /consumers/console-consumer-72909/offsets/t0425/26 #这个6就是刚才新生产消费的消息条数cZxid = 0x2000000eactime = Fri Aug 23 10:14:09 CST 2019mZxid = 0x2000000efmtime = Fri Aug 23 10:33:09 CST 2019pZxid = 0x2000000eacversion = 0dataVersion = 1aclVersion = 0ephemeralOwner = 0x0dataLength = 1numChildren = 0结论：生产消息时，每隔10分钟会random一次partition，也就是隔10就随机选一次partition
三、leader 均衡机制
[root@spark1 bin]# ./kafka-topics.sh --zookeeper spark1:2181,spark2,spark3 --describe --topic t0425Topic:t0425 PartitionCount:3 ReplicationFactor:3 Configs: Topic: t0425 Partition: 0 Leader: 0 Replicas: 0,2,1 Isr: 0,2,1 Topic: t0425 Partition: 1 Leader: 1 Replicas: 1,0,2 Isr: 1,0,2 Topic: t0425 Partition: 2 Leader: 2 Replicas: 2,1,0 Isr: 2,1,0##PartitionCount:3 //分区数ReplicationFactor:3 //副本数##0号partition归0号leader管理，Replicas：副本位于哪些节点上，lsr：启动集群时，去哪些节点检查数据完整性Partition: 0 Leader: 0 Replicas: 0,2,1 Isr: 0,2,1Partition: 1 Leader: 1 Replicas: 1,0,2 Isr: 1,0,2Partition: 2 Leader: 2 Replicas: 2,1,0 Isr: 2,1,0##均衡机制一个Partition只能归一个leader管，但一个leader可以有多个Partition；当一个leader挂了，Partition会依赖副本优先机制，重新寻找一个leader；##副本优先机制如Partition:0 ->Partition: 0 Leader: 0 Replicas: 0,2,1partition0的副本位于Replicas: 0,2,1 ，也就是顺序为0 2 1，所以它就会按这个循序找leader；##再均衡当把挂掉的leader主机，重新起来以后，kafka会自动释放掉partition后来寻找的那个leader，重新恢复成原来的模样；
四、Kafka生产过程分析
1、写入方式
producer采用推（push）模式将消息发布到broker，每条消息都被追加（append）到分区（patition）中，属于顺序写磁盘（顺序写磁盘效率比随机写内存要高，保障kafka吞吐率）
2、partition(分区)
消息发送时都被发送到一个topic，其本质就是一个目录，而topic是由一些Partition Logs(分区日志)组成；每个Partition中的消息都是有序的，生产的消息被不断追加到Partition log上，其中的每一个消息都被赋予了一个唯一的offset值；分区的原因：方便在集群中扩展，每个Partition可以通过调整以适应它所在的机器，而一个topic又可以有多个Partition组成，因此整个集群就可以适应任意大小的数据了；可以提高并发，因为可以以Partition为单位读写了；分区的原则：指定了patition，则直接使用；未指定patition但指定key，通过对key的value进行hash出一个patition；patition和key都未指定，使用轮询选出一个patition；
3、Replication(副本 )
同一个partition可能会有多个replication（对应 server.properties 配置中的 default.replication.factor=N N是一个数字）。没有replication的情况下，一旦broker 宕机，其上所有 patition 的数据都不可被消费，同时producer也不能再将数据存于其上的patition。引入replication之后，同一个partition可能会有多个replication，而这时需要在这些replication之间选出一个leader，producer和consumer只与这个leader交互，其它replication作为follower从leader 中复制数据；
4、写入流程
1、producer先从zookeeper的 "/brokers/.../state"节点找到该partition的leader2、producer将消息发送给该leader3、leader将消息写入本地log4、followers从leader pull消息5、写入本地log后向leader发送ACK6、leader收到所有ISR中的replication的ACK后，增加HW（high watermark，最后commit 的offset）并向producer发送ACK
5、broker消息存储
##存储方式物理上把topic分成一个或多个patition（对应 server.properties 中的num.partitions=N配置，N是一个数字），每个patition物理上对应一个文件夹（该文件夹存储该patition的所有消息和索引文件），如下：[root@spark1 bin]# ./kafka-topics.sh --zookeeper spark1:2181,spark2,spark3 --create --topic t0425 --partitions 3 --replication-factor 3[root@spark1 kafka-logs]# ls |grep t0425 #在kafka的数据目录中t0425-0t0425-1t0425-2[root@spark1 kafka-logs]# ll t0425-0总用量 0-rw-r--r-- 1 root root 10485760 8月 23 11:41 00000000000000000000.index-rw-r--r-- 1 root root 0 8月 23 11:41 00000000000000000000.log[root@spark1 kafka-logs]# ll t0425-1总用量 0-rw-r--r-- 1 root root 10485760 8月 23 11:41 00000000000000000000.index-rw-r--r-- 1 root root 0 8月 23 11:41 00000000000000000000.log[root@spark1 kafka-logs]# ll t0425-2总用量 0-rw-r--r-- 1 root root 10485760 8月 23 11:41 00000000000000000000.index-rw-r--r-- 1 root root 0 8月 23 11:41 00000000000000000000.log##存储策略无论消息是否被消费，kafka都会保留所有消息。有两种策略可以删除旧数据基于时间：log.retention.hours=168基于大小：log.retention.bytes=1073741824需要注意的是，因为Kafka读取特定消息的时间复杂度为O(1)，即与文件大小无关，所以这里删除过期文件与提高 Kafka 性能无关；
6、zookeeper存储结构
五、Kafka消费过程分析
kafka提供了两套consumer API，高级Consumer API和低级Consumer API
1、高级API
高级API优点： 高级API 写起来简单 不需要自行去管理offset，系统通过zookeeper自行管理 不需要管理分区，副本等情况，.系统自动管理 消费者断线会自动根据上一次记录在zookeeper中的offset去接着获取数据（默认设置1分钟更新一下zookeeper中存的offset） 可以使用group来区分对同一个topic 的不同程序访问分离开来（不同的group记录不同的offset，这样不同程序读取同一个topic才不会因为offset互相影响）高级API缺点： 不能自行控制offset（对于某些特殊需求来说） 不能细化控制如分区、副本、zk等，不关心数据会不会丢失。
2、低级API
低级 API 优点 能够让开发者自己控制offset，想从哪里读取就从哪里读取。 自行控制连接分区，对分区自定义进行负载均衡 对zookeeper的依赖性降低（如：offset不一定非要靠zk存储，自行存储offset即可，比如存在文件或者内存中）低级API缺点 太过复杂，需要自行控制offset，连接哪个分区，找到分区leader 等
3、消费者组
消费者是以consumer group消费者组的方式工作，由一个或者多个消费者组成一个组，共同消费一个topic。每个分区在同一时间只能由group中的一个消费者读取，但是多个group可以同时消费这个partition。在上图中，有一个由三个消费者组成的group，有一个消费者读取主题中的两个分区，另外两个分别读取一个分区。某个消费者读取某个分区，也可以叫做某个消费者是某个分区的拥有者。在这种情况下，消费者可通过水平扩展的方式同时读取大量的消息。另外，如果一个消费者失败了，那么其他的group成员会自动负载均衡读取之前失败的消费者读取的分区；
4、消费方式
consumer采用pull（拉）模式从broker中读取数据。push（推）模式很难适应消费速率不同的消费者，因为消息发送速率是由broker决定的。它的目标是尽可能以最快速度传递消息，但是这样很容易造成consumer来不及处理消息，典型的表现就是拒绝服务以及网络拥塞。而pull模式则可以根据consumer的消费能力以适当的速率消费消息。对于Kafka而言，pull模式更合适，它可简化broker的设计，consumer可自主控制消费消息的速率，同时consumer可以自己控制消费方式——即可批量消费也可逐条消费，同时还能选择不同的提交方式从而实现不同的传输语义。pull模式不足之处是，如果kafka没有数据，消费者可能会陷入循环中，一直等待数据到达。为了避免这种情况，我们在我们的拉请求中有参数，允许消费者请求在等待数据到达的“长轮询”中进行阻塞（并且可选地等待到给定的字节数，以确保大的传输大小）；