3、kafka工作流程

kafka&＃xff1a;分布式消息系统&＃xff0c;将消息直接存入磁盘&＃xff0c;默认保存一周。broker&＃xff1a;组成kafka集群的节点&＃xff0c;之间没有主从关系&＃xff0c;依赖zookeeper来协调&＃xff0c;broker负责满息的读写和存储&＃xff0c;一个broker可以管理多个partition.producer&＃xff1a;消息的生产者&＃xff0c;自己决定向哪个partition中去生产消息&＃xff0c;两种机利&＃xff1a;hash&＃xff0c;轮循。consumer&＃xff1a;消息的消费者&＃xff0c;consumer通过zookeeper去维护消费者偏移量。consumer有自己的消费者组&＃xff0c;不同的组之间消费同一个topic数据&＃xff0c;互不影响&＃xff0c;相同的组内的不同的consumer消费同一个topic&＃xff0c;这个topic中相同的数据只能被消费一次。topic&＃xff1a;一类消息总称/一个消息队列。topic是由partition组成的&＃xff0c;有多少&＃xff1f;-> 创建指定。partition&＃xff1a;组成topic的单元&＃xff0c;每个partition有副本&＃xff0c;有多少&＃xff1f;-> 创建topic时指定。每个partition只能由一个broker来管理&＃xff0c;这个broker就是这个partition的leader。zookeeper&＃xff1a;协调kafka broker&＃xff0c;存储原数据&＃xff1a;consumer的offset&＃43;broker信息&＃43;topic信息&＃43;partition信息&＃xff1b;

[root&＃64;spark1 bin]# pwd
/usr/local/kafka/bin#创建topic
[root&＃64;spark1 bin]# ./kafka-topics.sh --zookeeper spark1:2181,spark2,spark3 --create --topic t0425 --partitions 3 --replication-factor 3
Created topic "t0425".[root&＃64;spark1 bin]# ./kafka-topics.sh --zookeeper spark1:2181,spark2,spark3 --list |grep t0425
t0425

##生产者
[root&＃64;spark1 bin]# ./kafka-console-producer.sh --topic t0425 --broker-list spark1:9092,spark2:9092,spark3:9092
hello
world
1
2
3
...##消费者
[root&＃64;spark1 bin]# ./kafka-console-consumer.sh --zookeeper spark1:2181,spark2:2181,spark3:2181 --topic t0425
hello
world
1
2
3
...

##连接zookeeper
[root&＃64;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&＃xff0c;也就是刚才创建的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 &＃61; 0x2000000ce
ctime &＃61; Fri Aug 23 10:06:02 CST 2019
mZxid &＃61; 0x2000000ce
mtime &＃61; Fri Aug 23 10:06:02 CST 2019
pZxid &＃61; 0x2000000ce
cversion &＃61; 0
dataVersion &＃61; 0
aclVersion &＃61; 0
ephemeralOwner &＃61; 0x0
dataLength &＃61; 76
numChildren &＃61; 0

[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/0
12 #这个12就是刚才生产消费的消息条数&＃xff0c;都在0里面,1和2里面没有
cZxid &＃61; 0x2000000e7
ctime &＃61; Fri Aug 23 10:14:09 CST 2019
mZxid &＃61; 0x2000000e7
mtime &＃61; Fri Aug 23 10:14:09 CST 2019
pZxid &＃61; 0x2000000e7
cversion &＃61; 0
dataVersion &＃61; 0
aclVersion &＃61; 0
ephemeralOwner &＃61; 0x0
dataLength &＃61; 2
numChildren &＃61; 0##现在接着去生产几条消息&＃xff0c;然后再看0 1 2
[zk: localhost:2181(CONNECTED) 18] get /consumers/console-consumer-72909/offsets/t0425/2
6 #这个6就是刚才新生产消费的消息条数
cZxid &＃61; 0x2000000ea
ctime &＃61; Fri Aug 23 10:14:09 CST 2019
mZxid &＃61; 0x2000000ef
mtime &＃61; Fri Aug 23 10:33:09 CST 2019
pZxid &＃61; 0x2000000ea
cversion &＃61; 0
dataVersion &＃61; 1
aclVersion &＃61; 0
ephemeralOwner &＃61; 0x0
dataLength &＃61; 1
numChildren &＃61; 0结论&＃xff1a;生产消息时&＃xff0c;每隔10分钟会random一次partition&＃xff0c;也就是隔10就随机选一次partition

[root&＃64;spark1 bin]# ./kafka-topics.sh --zookeeper spark1:2181,spark2,spark3 --describe --topic t0425
Topic:t0425 PartitionCount:3 ReplicationFactor:3 Configs:Topic: t0425 Partition: 0 Leader: 0 Replicas: 0,2,1 Isr: 0,2,1Topic: t0425 Partition: 1 Leader: 1 Replicas: 1,0,2 Isr: 1,0,2Topic: t0425 Partition: 2 Leader: 2 Replicas: 2,1,0 Isr: 2,1,0##
PartitionCount:3 //分区数
ReplicationFactor:3 //副本数

##0号partition归0号leader管理&＃xff0c;Replicas&＃xff1a;副本位于哪些节点上&＃xff0c;lsr&＃xff1a;启动集群时&＃xff0c;去哪些节点检查数据完整性
Partition: 0 Leader: 0 Replicas: 0,2,1 Isr: 0,2,1
Partition: 1 Leader: 1 Replicas: 1,0,2 Isr: 1,0,2
Partition: 2 Leader: 2 Replicas: 2,1,0 Isr: 2,1,0##均衡机制
一个Partition只能归一个leader管&＃xff0c;但一个leader可以有多个Partition&＃xff1b;
当一个leader挂了&＃xff0c;Partition会依赖副本优先机制&＃xff0c;重新寻找一个leader&＃xff1b;##副本优先机制
如Partition:0 ->Partition: 0 Leader: 0 Replicas: 0,2,1
partition0的副本位于Replicas: 0,2,1 &＃xff0c;也就是顺序为0 2 1&＃xff0c;所以它就会按这个循序找leader&＃xff1b;##再均衡
当把挂掉的leader主机&＃xff0c;重新起来以后&＃xff0c;kafka会自动释放掉partition后来寻找的那个leader&＃xff0c;重新恢复成原来的模样&＃xff1b;

producer采用推&＃xff08;push&＃xff09;模式将消息发布到broker&＃xff0c;每条消息都被追加&＃xff08;append&＃xff09;到分区&＃xff08;patition&＃xff09;中&＃xff0c;属于顺序写磁盘&＃xff08;顺序写磁盘效率比随机写内存要高&＃xff0c;保障kafka吞吐率&＃xff09;

消息发送时都被发送到一个topic&＃xff0c;其本质就是一个目录&＃xff0c;而topic是由一些Partition Logs(分区日志)组成&＃xff1b;每个Partition中的消息都是有序的&＃xff0c;生产的消息被不断追加到Partition log上&＃xff0c;其中的每一个消息都被赋予了一个唯一的offset值&＃xff1b;分区的原因&＃xff1a;
方便在集群中扩展&＃xff0c;每个Partition可以通过调整以适应它所在的机器&＃xff0c;而一个topic又可以有多个Partition组成&＃xff0c;因此整个集群就可以适应任意大小的数据了&＃xff1b;可以提高并发&＃xff0c;因为可以以Partition为单位读写了&＃xff1b;分区的原则&＃xff1a;
指定了patition&＃xff0c;则直接使用&＃xff1b;未指定patition但指定key&＃xff0c;通过对key的value进行hash出一个patition&＃xff1b;patition和key都未指定&＃xff0c;使用轮询选出一个patition&＃xff1b;

同一个partition可能会有多个replication&＃xff08;对应 server.properties 配置中的 default.replication.factor&＃61;N N是一个数字&＃xff09;。没有replication的情况下&＃xff0c;一旦broker 宕机&＃xff0c;
其上所有 patition 的数据都不可被消费&＃xff0c;同时producer也不能再将数据存于其上的patition。引入replication之后&＃xff0c;同一个partition可能会有多个replication&＃xff0c;
而这时需要在这些replication之间选出一个leader&＃xff0c;producer和consumer只与这个leader交互&＃xff0c;其它replication作为follower从leader 中复制数据&＃xff1b;

1、producer先从zookeeper的 "/brokers/.../state"节点找到该partition的leader
2、producer将消息发送给该leader
3、leader将消息写入本地log
4、followers从leader pull消息
5、写入本地log后向leader发送ACK
6、leader收到所有ISR中的replication的ACK后&＃xff0c;增加HW&＃xff08;high watermark&＃xff0c;最后commit 的offset&＃xff09;并向producer发送ACK

##存储方式
物理上把topic分成一个或多个patition&＃xff08;对应 server.properties 中的num.partitions&＃61;N配置&＃xff0c;N是一个数字&＃xff09;&＃xff0c;
每个patition物理上对应一个文件夹&＃xff08;该文件夹存储该patition的所有消息和索引文件&＃xff09;&＃xff0c;如下&＃xff1a;[root&＃64;spark1 bin]# ./kafka-topics.sh --zookeeper spark1:2181,spark2,spark3 --create --topic t0425 --partitions 3 --replication-factor 3[root&＃64;spark1 kafka-logs]# ls |grep t0425 #在kafka的数据目录中
t0425-0
t0425-1
t0425-2[root&＃64;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&＃64;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&＃64;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##存储策略
无论消息是否被消费&＃xff0c;kafka都会保留所有消息。有两种策略可以删除旧数据基于时间&＃xff1a;log.retention.hours&＃61;168
基于大小&＃xff1a;log.retention.bytes&＃61;1073741824需要注意的是&＃xff0c;因为Kafka读取特定消息的时间复杂度为O(1)&＃xff0c;即与文件大小无关&＃xff0c;所以这里删除过期文件与提高 Kafka 性能无关&＃xff1b;

kafka提供了两套consumer API&＃xff0c;高级Consumer API和低级Consumer API

高级API优点&＃xff1a;高级API 写起来简单不需要自行去管理offset&＃xff0c;系统通过zookeeper自行管理不需要管理分区&＃xff0c;副本等情况&＃xff0c;.系统自动管理消费者断线会自动根据上一次记录在zookeeper中的offset去接着获取数据&＃xff08;默认设置1分钟更新一下zookeeper中存的offset&＃xff09;可以使用group来区分对同一个topic 的不同程序访问分离开来&＃xff08;不同的group记录不同的offset&＃xff0c;这样不同程序读取同一个topic才不会因为offset互相影响&＃xff09;高级API缺点&＃xff1a;不能自行控制offset&＃xff08;对于某些特殊需求来说&＃xff09;不能细化控制如分区、副本、zk等&＃xff0c;不关心数据会不会丢失。

低级 API 优点能够让开发者自己控制offset&＃xff0c;想从哪里读取就从哪里读取。自行控制连接分区&＃xff0c;对分区自定义进行负载均衡对zookeeper的依赖性降低&＃xff08;如&＃xff1a;offset不一定非要靠zk存储&＃xff0c;自行存储offset即可&＃xff0c;比如存在文件或者内存中&＃xff09;低级API缺点太过复杂&＃xff0c;需要自行控制offset&＃xff0c;连接哪个分区&＃xff0c;找到分区leader 等

消费者是以consumer group消费者组的方式工作&＃xff0c;由一个或者多个消费者组成一个组&＃xff0c;共同消费一个topic。每个分区在同一时间只能由group中的一个消费者读取&＃xff0c;
但是多个group可以同时消费这个partition。在上图中&＃xff0c;有一个由三个消费者组成的group&＃xff0c;有一个消费者读取主题中的两个分区&＃xff0c;另外两个分别读取一个分区。
某个消费者读取某个分区&＃xff0c;也可以叫做某个消费者是某个分区的拥有者。在这种情况下&＃xff0c;消费者可通过水平扩展的方式同时读取大量的消息。另外&＃xff0c;如果一个消费者失败了&＃xff0c;那么其他的group成员会自动负载均衡读取之前失败的消费者读取的分区&＃xff1b;

consumer采用pull&＃xff08;拉&＃xff09;模式从broker中读取数据。push&＃xff08;推&＃xff09;模式很难适应消费速率不同的消费者&＃xff0c;因为消息发送速率是由broker决定的。它的目标是尽可能以最快速度传递消息&＃xff0c;但是这样很容易造成consumer来不及
处理消息&＃xff0c;典型的表现就是拒绝服务以及网络拥塞。而pull模式则可以根据consumer的消费能力以适当的速率消费消息。对于Kafka而言&＃xff0c;pull模式更合适&＃xff0c;它可简化broker的设计&＃xff0c;consumer可自主控制消费消息的速率&＃xff0c;同时consumer可以自己控制消费方式——即可批量消费也可逐条消费&＃xff0c;
同时还能选择不同的提交方式从而实现不同的传输语义。pull模式不足之处是&＃xff0c;如果kafka没有数据&＃xff0c;消费者可能会陷入循环中&＃xff0c;一直等待数据到达。为了避免这种情况&＃xff0c;我们在我们的拉请求中有参数&＃xff0c;允许消费者请求在等待
数据到达的“长轮询”中进行阻塞&＃xff08;并且可选地等待到给定的字节数&＃xff0c;以确保大的传输大小&＃xff09;&＃xff1b;