- 1. 什么是分布式？
- 2. 什么是Kafka ？
- 3. 为什么用Kafka （使用消息队列的好处）？
- 4. Kafka的组成（架构） ？
- 5. Kafka的副本（Replication）？
- 6. 如何增加消费能力？
- 7. Kafka为什么读写效率高（高吞吐） ？
- 8. Kafka命令行操作
- 9. Kafka写入方式？
- 10. Kafka写入数据（生产）的流程 ？
- 11. Kafka分区策略 ？
- 12. Kafka的选举Leader机制 ？
- 未完待续...
1.1. 用户分布式
有多个用户向服务器发起请求，一个请求就是一个线程，线程要分配栈空间，空间不够则增加服务器数量；
而用户请求的端口号是唯一的，多个服务器的端口号不一样，所以需要负载均衡器用来调度，用户向负载均衡器发起请求，负载均衡器根据服务器状态选择服务器；
而多个服务器之间的数据需要通信，进程是独立的，进程通信性能差，所以使用缓存服务器，将数据都存在缓存服务器，然后共享里面的数据，在内存中，性能好。
由于数据传输也要经过负载均衡器，使其负载大，所以使用注册中心Zookeeper；
Zookeeper存入服务器的ip和端口号；
①用户先访问Zookeeper，Zookeeper将ip和端口号给用户，②由用户轮询选择服务器，让负载由用户区承担，使Zookeeper负荷小，性能提高，然后用户直接连接服务器；
1.2. 分布式系统
把一个完整的系统拆分成了若干个子系统来分别运行-------分布式系统
如京东购物，有订单、管理、物流、财务等子系统；
当其中订单子系统的负载大，由于使用了同一网络，网络被订单子系统占用，其他服务器也会被影响，所以不同的子系统放在不同的机器和网络中；
而不同的子系统直接交互会导致耦合，所以这时使用消息队列----Kafka
子系统会将数据的变化放到消息队列中，分类管理起来，不同的服务器系统周期性的去消息队列中取消息，每个子系统只从消息队列中获取自己需要的数据；
Kafka是一种高吞吐量、分布式、基于发布/订阅的消息系统；Kafka对消息保存时根据Topic进行归类，发送消息者称为Producer生产者，消息接受者称为Consumer消费者，此外kafka集群有多个kafka服务器组成，服务器节点称为broker；
类似一个存放数据的中间商，和百度网盘差不多，你往百度网盘上传一个文件（produce），然后发布链接（topic），别人可以通过你的链接，下载文件；
- 解耦合： 将生产、消费、数据存储分开，耦合性低、扩展性强（比较容易添加机器）；
- 灵活性&峰值处理能力： 访问量剧增时，使用消息队列可以使数据存着让消费者来取就可以，避免因为剧增的负荷使系统崩溃；
- 可恢复性： 当服务器宕机时会有副本继续提供服务，当宕机的服务器恢复，会继续使用；
- 异步通信： 当发一条消息后不需要等待结果可以做其他操作就是异步 ；而发了消息需要等待结果才能继续操作就是同步，用户可以把消息放入队列而不处理，需要的时候再去处理；
- 缓存： 提供缓冲，解决生产消息和消费消息的处理速度不一致的情况
Kafka集群的目的是保存消息，让Consumer去消费数据，消息来自于Producer，生产者把数据以K、V的方式传给集群；
- 首先Kafka需要多台机器组成一个Kafka集群才能承载符合，每一台服务器是一个Broker，
- 数据有多种进行分类 → 一个Broker中有Topic主题
- 为了提高负载 → 一个Topic有多个partition分区，一个partition可以放在多个不同的Broker上，而数据放在不同的partition当中（取模的方式），可以让不同的消费者来消费，提高消费速率
- 为了防止数据丢失 → 副本备份Leader和Follower，Leader做io处理，Follower只作备份，多个分区进行“交叉备份”，Follower从Leader中实时同步数据，当Leader挂掉后吗，Follower会代替Leader。
以消费者组为单位进行读取数据，其中一个消费者读取一个partiion分区，所以一般partition分区数量=消费者组中的Consumer消费者数量
Zookeeper：kafka集群依赖zookeeper来保存集群的的元信息，来保证系统的可用性。
Kafka -> Broker -> Topic -> partition -> Replication(Leader、Follower) ->Consumer
1）Producer ：消息生产者，就是向 kafka broker 发消息的客户端；
2）Consumer ：消息消费者，向 kafka broker 取消息的客户端；
3）Consumer Group （CG）：消费者组，由多个 consumer 组成。消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个组内消费者消费；消费者组之间互不影响。
所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。
4）Broker ：一台 kafka 服务器就是一个 broker。一个集群由多个 broker 组成。
一个 broker可以容纳多个topic。
5）Topic ：可以理解为一个队列，生产者和消费者面向的都是一个 topic；topic逻辑上的概念，partition是物理上的概念；
6）Partition：为了实现扩展性，一个非常大的 topic 可以分布到多个broker（即服务器）上，一个 topic 可以分为多个 partition，每个 partition 是一个有序的队列；
7）Replica：副本，为保证集群中的某个节点发生故障时，该节点上的 partition 数据不丢失，
且kafka 仍然能够继续工作，kafka 提供了副本机制，一个 topic 的每个分区都有若干个副本，一个 leader 和若干个 follower。
8）leader：每个分区多个副本的“主”，生产者发送数据的对象，以及消费者消费数据的对
象都是 leader。
9）follower：每个分区多个副本中的“从”，实时从 leader 中同步数据，保持和 leader 数据的同步。leader 发生故障时，某个 follower 会成为新的 follower。
10）Offset：偏移量， kafka的存储文件都是按照offset.kafka来命名，用offset做名字的好处是方便查找。例如你想找位于2049的位置，只要找到2048.kafka的文件即可。当然the first offset就是00000000000.kafka。
同一个partition分区可能会有多个replication（对应 server.properties 配置中的 default.replication.factor=N）。
Leader和follower都算副本；
没有replication的情况下，一旦broker 宕机，其上所有 patition 的数据都不可被消费，同时producer也不能再将数据存于其上的patition。
引入replication之后，同一个partition可能会有多个replication备份，而这时需要在这些replication之间选出一个leader，producer和consumer只与这个leader交互，其它replication作为follower从leader 中复制数据（同步数据）。
同步数据：Leader负责数据的读和写，如果让Leader再往Follower发送数据则会使Leader负载太大，所以由Follower轮询去Leader中取数据 ！
所以，Producer对Leader做操作，而Follower定时向Leader取数据以此备份；
由于消费者数量一般等于partition分区数量，所以增加partition分区可以增加消费能力；
如果分区不增加只增加消费者，则多出来的消费者会被浪费，因为一个分区只能被一个消费者组内的一个消费者进行消费；
首先Kafka是分布式集群，吞吐量大；
-
零拷贝
实际应用中，如果需要把磁盘的内容发送到远程服务器：
1.操作系统将磁盘的数据A复制到内核缓存，
2.再把缓存数据复制到应用缓存中，
3.在应用中使用write()方法把应用缓存中数据复制到PageCache内核缓存中，
4.再由系统将内核缓存的数据传给NIC Buffer网卡缓冲区，通过网卡发送给数据接收方。
而Kafka：
直接和底层操作系统打交道，向系统发送sendFile指令（操作系统级别的指令），让数据从磁盘拷贝到内核缓存，然后直接发给NIC 网卡缓冲区，而不需要再通过应用层；
所谓的零拷贝就是相对用户空间来说，不需要再进行数据的拷贝； -
顺写
数据是以日志文件的方式顺序保存 ，比随机写效率高； -
预读
读数据时会预先读相关的数据，提高效率（由空间局部性原理） -
后写
（先放入磁盘Cache缓存（访问缓存的速度比访问文件快很多），到达一定值再由操作系统把缓存的值写入文件） -
分段日志
（数据有offset偏移量，读取快）
index 和 log 文件以当前 segment 的第一条消息的 offset 命名；
Topic会分区，分区文件中存多个segment，一个segment有三种类型文件：
index 和 log 文件以当前 segment 的第一条消息的 offset 命名；
xxxxx.index 索引，存数据和在当前文件内数据的偏移量，用offeset去定位数据;
xxxxxx.log是一个分段日志，存的是数据，数据文件就是日志文件，默认日志文件大小1G，分为多个分段日志可以提高性能，避免大文件加载性能差；
xxxxxx.timeindex 时间戳索引，用时间去定位数据 -
批处理
（一批数据一块处理） -
双端队列DQ
每一个分区都会有一个双端队列DQ -
压缩：给的字符串，会被压缩成byte数组，压缩后数据小，传输快
1）查看当前服务器中的所有topic： bin/kafka-topics.sh --zookeeper hadoop102:2181 --list
–list意为查看列表
集群由Zookeeper来管理，要进行Topic操作，则集群需要先要连接上Zookeeper；
2181是Zookeeper的服务端口！
2）创建topic ：bin/kafka-topics.sh --zookeeper hadoop102:2181 --create --replication-factor 3 --partitions 3 --topic first
创建Topic的目的是为了能向某一个Topic主题中发送给消息
需要提供分区、副本（Leader、Follower都是副本）
–topic 定义topic名
–replication-factor 定义副本数
–partitions 定义分区数
3）查看当前Topic( first )的描述信息：bin/kafka-topics.sh --zookeeper hadoop102:2181 --describe --topic first
可以看到分区数partitionCount=3，副本数为2(一个leader一个follower)
意为0号分区的Leader在Broker 1号机器，即hadoop102这台虚拟机上
4）删除topic ：bin/kafka-topics.sh --zookeeper hadoop102:2181 --delete --topic first
5）发送消息（生产）： bin/kafka-console-producer.sh --broker-list hadoop102:9092 --topic first (打开生产者控制台)
>hello
>world
集群和消费跟Zookeeper有关系，但是生产和Zookeeper没有关系，只需要找到集群就够了；
这里使用控制台发送消息，所以要先启动控制台连接Kafka集群
9092是集群的服务端口！
6）消费消息: （打开消费者控制台）
bin/kafka-console-consumer.sh --zookeeper hadoop102:2181 --topic first
bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
消费和Zookeeper有关，需要连接Zookeeper，使用Zookeeper来保存offset;
**–from-beginning：**会把first主题中以往所有的数据都读取出来，即从头读;
producer采用推（push）模式将消息发布到broker，每条消息都被追加（append）到patition分区的分段日志log末端，保证分区内有序，属于顺序写磁盘（顺序写磁盘效率比随机写内存要高，保障kafka吞吐率）。
由于生产者生产的消息会不断追加到 log 文件末尾，为防止 log 文件过大导致数据定位效率低下，Kafka 采取了分区和索引机制，将每个 partition 分为多个 segment。每个 segment 对应三个文件——“.index”文件、“timeindex”文件、 “.log”文件。这些文件位于一个文件夹下，该文件夹的命名规则为：topic 名称+分区序号。
例如，first 这个 topic 有三个分区，则其对应的文件夹为 first- 0,first-1,first-2。
index 、timeindex和 log 文件以当前 segment 的第一条消息的 offset 命名
集群依靠Zookeeper来管理；
Consumer消费者靠Zookeeper来保存偏移量；但是生产者与Zookeeper无关，Producer不和Zookeeper直接打交道；
过程：
1.Producer需要找集群并将数据放到某一个分区中，而分区和Leader端口号在Zookeeper中，集群和Zookeeper一直在连接，由集群去取Partition的Leader端口号，然后producer从集群中拿到对应的topic的partition信息和partition的leader的相关信息
2.Producer将数据发送给Leader，先放到双端队列DQ，DQ满了或者达到时间周期就由sender发送到Leader（数据采集器，双端队列DQ、sender，）
3.Leader将数据写入本地log分段日志
后续：
4.Follower轮询从Leader拉取消息以此同步数据
5.Kafka的ACK应答机制；
- 当取值为0，则不关心是否到达，尽最大努力交付，效率高，数据可能丢失；
- 取值为1（默认），Producer的发送数据，需要等待Leader的应答才能发生下一条，不关心Follower是否接收成功，性能稍慢，数据较安全，但当Leader突然宕机，则当Follower还未同步，数据会丢失；
- 取值为-1 ，Producer发送数据，需要等待所有副本（leader和所有Follower）都完成备份，最安全，性能差；
详细：
1.首先Producer要向集群的Leader发送消息，需要知道分区信息和Leader的端口号，需要先连接集群；使用Sender数据发送器发送请求（Sender实现了Runnable接口，是个独立的线程，Sender用来实现数据的交互），获取到集群Topic的partition和Leader信息；而集群和Zookeeper一直连接，Zookeeper有以上集群的信息，
2.当Producer通过Sender从集群获取到partition和Leader信息，若有指定partition则使用指定的partition，若没有则使用分区算法对key做操作；当没有key则轮询partition；
3.Producer给Leader发数据使用批处理，如果没有批处理每次发送都建立连接即在进程间做交互，会使效率很低；
这里使用了双端队列DQ，Producer将数据放入DQ，当DQ满了或者到达时间周期就由Sender（内部有采集器）取数据并一次性发给Leader（Sender会轮询DQ，看双端队列是否满了，或者达到时间周期）
4.Leader将数据写入本地log分段日志
后续：
5.Follower轮询从Leader拉取消息以此同步数据
6.Kafka的ACK应答机制；
为什么不让Producer直接和Zookeeper连接以获取集群信息？
因为这样会连接Zookeeper和集群共两次连接，而如果Producer连接集群以获取信息则只需要连接一次，性能提升。
为什么使用双端队列？
当从DQ发送到Sender失败之后，如果只是单向队列（先进先出）的DQ，则会从头放进去，这样就打破了顺序，而使用双端队列，可以将数据从队列末端再放进去，以保持分区内数据有序（分区数=DQ数），再尝试发送；
双端队列靠RecordAccumulator数据收集器来完成；
每一个Topic的partition分区都会创建一个DQ双端队列；
消息发送时都被发送到一个topic，其本质就是一个目录，而topic是由一些Partition Logs(分区日志)组成，每个分区都有自己的分区日志，其组织结
每个Partition分区中的消息都是有序的，生产的消息被不断追加到Partition log上，其中的每一个消息都被赋予了一个唯一的offset偏移量值。
1. 分区的原因
（1）方便在集群中扩展，每个Partition可以通过调整以适应它所在的机器，而一个topic又可以有多个Partition组成，因此整个集群就可以适应任意大小的数据了；数据大了多几个partition即可，即为了便于增加负载；
（2）可以提高并发，因为可以以Partition为单位读写了。
2. 分区的原则
（1）指定了patition，则直接使用；
（2）未指定patition但指定key，通过对key的value进行hash取余出一个patition；
（3）patition和key都未指定，使用轮询选出一个patition。
- 先选出【Zookeeper中的】/controller 集群控制器：
假设有3个Broker机器，而分区、再平衡都由其中一个broker也就是集群控制器来做；
在Zookeeper中有一个 临时节点 /controller，三个Broker都会向Zookeeper发请求，三个Broker都会创建 /controller临时节点，最先创建 /controller，谁就是 controller，此时会增加一个watch监听者，监听/controller节点有没有挂掉，一旦挂掉再次选举controller，谁先建立谁就是controller； - 当有了 controller集群控制器（Broker）有了之后，再选举Leader，会从ISR （正在同步的副本）中选择Leader，一般是列表中的第一个，
当副本全挂了就是-1，会等待有副本活过来；
