Kafka集群支持动态扩展，可弹性伸缩将生产集群规模扩大到上千个代理和数十万个分区。

将数据流安全地存储在分布式、持久、容错的集群中。

分布式、分区、复制和容错，支持数据备份，在可用性区域上有效地扩展集群或者跨地理区域连接独立的集群。

应用场景

Kafka作为一个消息中间件最基本的用途为解耦、异步、削峰、消息通讯，下面为其常用的场景：

• 日志收集：可以用Kafka可以收集各种服务的log，通过kafka以统一接口服务的方式开放给各种consumer。
• 消息系统：分布式发布、订阅消息系统，消息队列将消息生产者和订阅者分离，实现应用解耦和缓存消息等。
• 流量削峰：在应用前端以消息队列接收请求，当请求超过队列长度，直接不处理重定向至一个静态页面，来达到削峰的目的，比如用于秒杀活动，因为秒杀活动高度集中用户访问导致流量暴增，可能导致应用系统雪崩。
• 流式处理：结合主流的流式处理框架如Flink、Spark Streaming、Storm。
• 用户活动跟踪：kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后消费者通过订阅这些topic来做实时的监控分析，亦可保存到数据库。
• 运营指标：kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。

安装

我们这里规划部署192.168.50.34、192.168.50.35、192.168.50.36共3个节点的Kafka集群，当然是需要有基本JDK环境，Kafka部署还需依赖ZooKeeper，刚好上一篇文章我们也非常愉快的学习和部署Zookeeper集群，直接拿来使用，奥利给！

#官网下载
wget --no-check-certificate https://dlcdn.apache.org/kafka/3.0.0/kafka_2.13-3.0.0.tgz
#解压
tar -zxvf kafka_2.13-3.0.0.tgz
#进入kafka目录下创建logs文件夹
cd kafka_2.13-3.0.0
mkdir logs
#并修改config目录的server.properties
vim config/server.properties

修改server.properties配置文件内容如下：

// 作为当前机器在集群中的唯一标识，和zookeeper的myid性质一样
broker.id=1
// 监听地址和端口号，默认是9092
listeners=PLAINTEXT://192.168.50.34:9092
// 消息存放的目录
log.dirs=/home/commons/kafka_2.13-3.0.0/logs
// zookeeper集群地址
zookeeper.connect=192.168.50.34:2181,192.168.50.35:2181,192.168.50.36:2181

#然后将上整个kafka_2.13-3.0.0拷贝到另外两台192.168.50.35、192.168.50.36上的相同目录（我们这里是指/home/commons），配置文件主要修改broker.id和listeners，broker.id唯一那我们就顺序编号为2和3，listeners中的host就分配与主机地址对应即可，修改完成后我们可以在kafka根目录下分别启动三台kafka server，后台常驻方式带上参数 -daemon
./bin/kafka-server-start.sh -daemon config/server.properties
#启动后可以通过ps或者jps检查进程信息和logs下的日志文件如server.log
ps -ef | grep kafka
#停止kafka server
./bin/kafka-server-stop.sh

集群简单测试

• 创建、列出所有topic、查看指定的topic

#3.0版本官方使用-–bootstrap-server替代–-zookeeper，创建名称为mytopic1的Topic，指定分区数为1，分区的副本数为1
bin/kafka-topics.sh --create --bootstrap-server 192.168.50.34:9092,192.168.50.35:9092,192.168.50.36:9092 --replication-factor 1 --partitions 1 --topic mytopic1
#创建名称为mytopic2的Topic，指定分区数为2，分区的副本数为2
bin/kafka-topics.sh --create --bootstrap-server 192.168.50.34:9092,192.168.50.35:9092,192.168.50.36:9092 --replication-factor 2 --partitions 2 --topic mytopic2
#列出所有的topic
bin/kafka-topics.sh --list --bootstrap-server 192.168.50.34:9092,192.168.50.35:9092,192.168.50.36:9092
#查看指定topic
$bin/kafka-topics.sh --describe --bootstrap-server 192.168.50.34:9092,192.168.50.35:9092,192.168.50.36:9092 --topic mytopic1
$bin/kafka-topics.sh --describe --bootstrap-server 192.168.50.34:9092,192.168.50.35:9092,192.168.50.36:9092 --topic mytopic2

• 增加Topic的partition数量和查看指定查看 topic 指定分区 offset 的最大值或最小值和删除topic

#将mytopic1的分区数量扩充到5个
bin/kafka-topics.sh --alter --bootstrap-server 192.168.50.34:9092,192.168.50.35:9092,192.168.50.36:9092 --partitions 5 --topic mytopic1
#time 为 -1 时表示最大值，为 -2 时表示最小值：
./bin/kafka-run-class.sh kafka.tools.GetOffsetShell --topic mytopic1 --time -1 --broker-list 192.168.50.34:9092,192.168.50.35:9092,192.168.50.36:9092 --partitions 0

• 删除topic

#删除mytopic2
kafka-topics.sh --delete --bootstrap-server 192.168.50.34:9092,192.168.50.35:9092,192.168.50.36:9092 --topic mytopic2

• 生产和消费消息

#往mytopic1里生产消息，这里可以使用--broker-list也可以使用--bootstrap-server
./bin/kafka-console-producer.sh --broker-list 192.168.50.34:9092,192.168.50.35:9092,192.168.50.36:9092 --topic mytopic1
#从头开始消费消息
./bin/kafka-console-consumer.sh --bootstrap-server 192.168.50.34:9092,192.168.50.35:9092,192.168.50.36:9092 --topic mytopic1 --from-beginning
#从尾部开始消费消息,从尾部开始取数据，必需要指定分区：如果需要取指定个数消息可以加上如--max-messages 5的参数
./bin/kafka-console-consumer.sh --bootstrap-server 192.168.50.34:9092,192.168.50.35:9092,192.168.50.36:9092 --topic mytopic1 --offset latest --partition 1

发送多条消息，最终在该分区下收到分发的该分区下的消息

• 指定消费者组

./bin/kafka-console-consumer.sh --bootstrap-server 192.168.50.34:9092,192.168.50.35:9092,192.168.50.36:9092 --topic mytopic1 -group mygroup1 --from-beginning

• 查看消费者组列表

./bin/kafka-consumer-groups.sh --bootstrap-server 192.168.50.34:9092,192.168.50.35:9092,192.168.50.36:9092 --list

• 查看消费者组详情和其他

#消费者组详情./bin/kafka-consumer-groups.sh --bootstrap-server 192.168.50.34:9092,192.168.50.35:9092,192.168.50.36:9092 --group mygroup1 --describe#删除消费者组./bin/kafka-consumer-groups.sh --bootstrap-server 192.168.50.34:9092,192.168.50.35:9092,192.168.50.36:9092 --group mygroup1 --delete#平衡Leader，--partition:指定需要重新分配leader的partition编号./bin/kafka-leader-election.sh --bootstrap-server 192.168.50.34:9092,192.168.50.35:9092,192.168.50.36:9092 --topic mytopic1 --partition=2 --election-type preferred#此外还有kafka自带压测工具./bin/kafka-producer-perf-test.sh --topic mytopic1 --num-records 1000 --record-size 1 --throughput 1000 --producer-props bootstrap.servers=192.168.50.34:9092,192.168.50.35:9092,192.168.50.36:9092

架构原理面试宝典

Kafka的架构和组成

一个典型的 Kafka 体系架构包括若干 Producer、若干 Broker、若干 Consumer，以及一个 ZooKeeper 集群，如下图所示。其中 ZooKeeper 是 Kafka 用来负责集群元数据的管理、控制器的选举等操作的。Producer 将消息发送到 Broker，Broker 负责将收到的消息存储到磁盘中，而 Consumer 负责从 Broker 订阅并消费消息。

• Broker：服务代理节点，Kafka 集群包含一个或多个服务器，这种服务器被称为 broker。对于 Kafka 而言，Broker 可以简单地看作一个独立的 Kafka 服务节点或 Kafka 服务实例。大多数情况下也可以将 Broker 看作一台 Kafka 服务器，前提是这台服务器上只部署了一个 Kafka 实例。一个或多个 Broker 组成了一个 Kafka 集群。
• Topic：每条发布到 Kafka 集群的消息都有一个类别，这个类别被称为 Topic， Topic属于逻辑概念，也即是物理上不一样 Topic 的消息分开存储但逻辑上一个 Topic 的消息虽然保存于一个或多个 broker 上，但用户只需指定消息的 Topic 便可生产或消费数据而没必要关心数据存于何处。
• Partition：Partition 属于物理概念，Topic的分区，每一个 Topic 包含一个或多个 Partition，分区的作用是做负载，提高kafka的吞吐量，同一个topic在不同的分区的数据是不重复的，partition的表现形式就是一个一个的文件夹。分区主要实现扩展和提高并发，以partition作为读写单位，可以同时多个生产者消费者读写。
• Replication:每一个分区可以有一个主分区（Leader）和多个副本（Follower），副本的作用是做备份。当主分区故障的时候会选择一个副本成为新的Leader。在kafka中副本的数量不能大于Broker的数量，follower和leader绝对是在不同的机器，同一机器对同一个分区也只可能存放一个副本（包括自己）。
• Producer： 生产者，也就是发送消息的一方。生产者负责创建消息，然后将其投递到 Kafka 中，负责发布消息到 Kafka broker。
• Consumer：消费者，也就是接收消息的一方，向 Kafka broker 读取消息的客户端，消费者连接到 Kafka 上并接收消息，进而进行相应的业务逻辑处理。
• Consumer Group：每一个 Consumer 属于一个特定的 Consumer Group（可为每一个 Consumer 指定 group name，若不指定 group name 则属于默认的 group），我们可以将多个消费者组成一个消费者组，在kafka的设计中同一个分区的数据只能被消费者组中的某一个消费者消费。同一个消费者组的消费者可以消费同一个topic的不同分区的数据，这也是为了提高kafka的吞吐量。
• Message：每一条发送的消息主体，也即是消息内容。
• Zookeeper：kafka集群依赖zookeeper来保存集群的的元信息，来保证系统的可用性，管理broker上下线、Topic的分区和副本分配、leader选举、consumer动态扩缩容和触发负载均衡，维护消费者关系和每个partition分区的消费信息；broker启动将ip、端口注册存储Zookeeper，存储topic和partition。

Kafka的四个核心API

• Producer API：生产者API允许应用程序将一组记录发布到一个或多个Kafka Topic中。
• Consumer AIP：消费者API允许应用程序订阅一个或多个Topic，并处理向他们传输的记录流。
• Streams API：流API允许应用程序充当流处理器，从一个或者多个Topic中消费输入流，并将输出流生成为一个或多个输出主题，从而将输入流有效地转换为输出流。
• Connector API：连接器API允许构建和运行可重用的生产者或消费者，这些生产者或消费者将Kafka Topic连接到现有的应用程序或数据系统。例如：连接到关系数据库的连接器可以捕获对表的修改信息。

消息队列的模式？

• 点对点模式

  消息生产者发送消息到消息队列中，然后消息消费者从队列中取出并且消费消息，消息被消费后，队列中不在存储。所以消息消费者不可能消费到已经被消费的消息；队列支持存在多个消费者，但是对于一个消息而言，只会有一个消费者可以消费；如果想发给多个消费者，则需要多次发送该条消息。

发布/订阅模式（一对多，消费者消费数据之后不会清除消息）

消息生产者将消息发布到topic中，同时有多个消息消费者（订阅）消费该消息，和点对点的方式不同，发布到topic的消息会被所有的订阅者消费；但是消息队列不是存储系统数据保留是期限的，例如Kafka默认是7天；kafka就是这种模式的，分为两种方式 一种是是消费者去主动去消费（拉取）消息，而不是生产者推送消息给消费者。
• 另外一种就是生产者主动推送消息给消费者，类似公众号。

Kafka的存储模型

• Kafka中消息是以Topic进行分类的，生产者生产消息，消费者消费消息，都是面向Topic的。而Topic在物理上的存储是分区存储的，即按Partition分布式存储。
• 每个Partition中的数据又是顺序写入log文件中进行存储。这样会出现分区log文件过大，导致的读取性能下降的问题。所以Kafka将log文件切分成了segment，每个segment由 .log数据存储文件 和 .index索引文件 和 .timeindex文件组成。其中***.log用于存储消息本身的数据内容，.index存储消息在文件中的位置（包括消息的逻辑offset和物理存储offset），.timeindex**存储消息创建时间和对应逻辑地址的映射关系。详细的结构如下图所示：

• 每个log文件和index文件的命名就是 文件中起始数据的偏移量，一个segment中由index定位到对应log文件中执行数据的原理如下图：

• Kafka将分区拆分成多个段是为了控制存储的文件大小，如果整个分区只保存为一个文件，那随着分区里消息的增多，文件也将越来越大，最后不可控制。而如果每个消息都保存为一个文件，那文件数量又将变得巨大，同样容易失去控制。所以kafka采用段设计方式，控制了每个文件的大小和文件的数量。同时**可以很方便地通过操作系统mmap机制映射到内存中，提高写入和读取效率。**还有另一个好处是当系统要清除过期数据时可以直接将过期的段文件删除即可。但是这里也会有一个问题，如果每个消息都要在index文件中保存位置信息，那么index文件也很容易变得很大，这样又会减弱上文所说的好处。所以在kafka中，index设计为稀疏索引来降低index的文件大小，这样，index文件存储的实际内容为：**该段消息在消息队列中的相对offset和在log文件中的物理偏移量映射的稀疏记录。**那么多少条消息会在index中保存一条记录呢？这个可以通过系统配置来进行设置。索引记录固定为8个字节大小，分别为4个字节的相对offset(消息在partition中全局offset减去该segment的起始offset),4个字节的消息具体存储文件的物理偏移量。
• index文件中根据需要查找的offset根据保存起始偏移量(文件名)的相对偏移量，定位到log中数据真实的位置。（类似HBase中采用LSM日志结构合并树设计思想，顺序写入HFile，磁头寻址次数少，顺序读/写性能好），Kafka不会在消费者拉取完消息后马上就清理消息，而是会保存段文件一段时间，直到其过期再标记为可清理，由后台程序定期进行清理。这种机制使得消费者可以重复消费消息，满足更灵活的需求。

• 查找机制：建立在offset为有序的基础上，利用segment+有序offset+稀疏索引+二分查找+顺序查找等多种手段来高效的查找数据。
• 存储策略：无论消息是否被消费，kafka都会保存所有的消息。那对于旧数据有什么删除策略呢？ 基于时间，默认配置是168小时（7天）。
• 基于大小，默认配置是1073741824。

分区分配策略

如果Consumer组中Consumer数量>Topic Partition数量根据Random-Robin策略(实际是个轮询)或者Range(划分范围，一个范围内的给到一个消费者)对Partition进行分配。此后这个消费者组中的消费者消费的Partition就被定下来了。所以在实际的应用中，建议消费者组的consumer的数量与partition的数量一致。

Offset 维护

comsumer需要记录已经消费到的偏移量以便故障或者后续继续消费。在0.9版本后Kafka将这些信息都保存在一个内置的Topic(

__comsumer_offset

)，默认5s自动提交一次，而此前的版本是保存在ZK中的，这样改进目的是： 优化效率，减轻ZK压力。
• 可以自己实现偏移量维护。

Kafka发送数据的流程和消息结构？

• Producer在写入数据的时候永远的找leader，不会直接将数据写入follower，发送的流程如下图所示。

• 消息写入leader后，follower是主动的去leader进行同步的。producer采用push模式将数据发布到broker，每条消息追加到分区中，顺序写入磁盘，所以保证同一分区内的数据是有序的。
• Message结构：上面说到log文件就实际是存储message的地方，我们在producer往kafka写入的也是一条一条的message，将生产者发送数据封装为ProducerRecord对象，包括topic、partition、时间戳、key、value、header等；封装消息结构包含消息体、消息大小、offset、压缩类型等。 offset：offset是一个占8byte的有序id号，它可以唯一确定每条消息在parition内的位置。
• 消息大小：消息大小占用4byte，用于描述消息的大小。
• 消息体：消息体存放的是实际的消息数据（被压缩过），占用的空间根据具体的消息而异。

Kafka数据分区的目的和原则？

• 数据会写入到不同的分区，分区的主要目的是： 增加读写的吞吐量。
• 方便扩展，因为一个topic可以有多个partition，所以我们可以通过扩展机器去轻松的应对日益增长的数据量。
• 提高并发，以partition为读写单位，可以多个消费者同时消费数据，提高了消息的处理效率。

partition在写入的时候可以指定需要写入的partition，如果有指定，则写入对应的partition。
• 如果没有指定partition，但是设置了数据的key，则会根据key的值hash出一个partition。
• 如果既没指定partition，又没有设置key，则会通过random-robin算法轮询得到分区(第一个得到一个随机数，后续的在此基础上自增)。

Kafka如何保证消息不丢失或者可靠性？

为保证 Producer 发送的数据，能可靠地发送到指定的 Topic，Topic 的每个 Partition 收到 Producer 发送的数据后，都需要向 Producer 发送 ACK 确认收到，如果 Producer 收到 ACK，就会进行下一轮的发送，否则重新发送数据。

• 副本数据同步策略

  何时发送 ACK？确保有 Follower 与 Leader 同步完成，Leader 再发送 ACK，这样才能保证 Leader 挂掉之后，能在 Follower 中选举出新的 Leader 而不丢数据。
• 多少个 Follower 同步完成后发送 ACK？全部 Follower 同步完成，再发送 ACK。

ISR机制

如果采用第二种方案，所有 Follower 完成同步，Producer 才能继续发送数据，设想有一个 Follower 因为某种原因出现故障，那 Leader 就要一直等到它完成同步。这个问题怎么解决？ Leader维护了一个动态的 in-sync replica set（ISR，同步副本的作用），只需要这个列表的中的follower和leader同步，当ISR中的follower完成数据的同步之后，leader就会给生产者发送ack。
• 当 ISR 集合中的 Follower 完成数据的同步之后，Leader 就会给 Follower 发送 ACK。
• 如果 Follower 长时间未向 Leader 同步数据，则该 Follower 将被踢出 ISR 集合，该时间阈值由 replica.lag.time.max.ms 参数设定。Leader 发生故障后，就会从 ISR 中选举出新的 Leader。
• Leader和follower（ISR）落盘才会返回ack，会有数据重复现象，如果在leader已经写完成，且follower同步完成，但是在返回ack的出现故障，则会出现数据重复现象；极限情况下，这个也会有数据丢失的情况，比如follower和leader通信都很慢，所以ISR中只有一个leader节点，这个时候，leader完成落盘，就会返回ack，如果此时leader故障后，就会导致丢失数据

ACK 应答机制

对于某些不太重要的数据，对数据的可靠性要求不是很高，能够容忍数据的少量丢失，所以没必要等 ISR 中的 Follower 全部接受成功。
• 为保证生产者发送的数据能可靠的发送到指定的topic，Kafka在Topic的每个partition收到生产者发送的数据后，都需要向生产者发送ack（确认收到），如果生产者收到ack，就会进行下一轮的发送，否则重新发送数据。也即是核心是通过ACK应答机制，在生产者向队列写入数据的时候可以设置参数来确认是否kafka接收到数据，这个参数可设置的值为如下: 0代表producer往集群发送数据不需要等到集群的返回，不确保消息发送成功。安全性最低但是效率最高。
• 1代表producer往集群发送数据只要leader应答就可以发送下一条，只确保leader发送成功。
• -1代表all代表producer往集群发送数据需要所有的follower都完成从leader的同步才会发送下一条，确保leader发送成功和所有的副本都完成备份。安全性最高，但是效率最低，这里的all是指ISR列表中。

如果往不存在的topic写数据，kafka会自动创建topic，分区和副本的数量根据默认配置都是1。

生产者：同步发送，或者通过发送的callback来实现，producer.send(msg,callback)。

消费者：需要考虑先更新offset还是先做消费处理，先做消费则可能引出重复消费。

注意：Kakfa事务回滚不会直接去删除消息，而是将消息对Consumer不可见。

Consumer事务

Kafka对Consumer的事务较弱，一般是通过Consumer端自己实现精确一次性消费(将消费过程和提交offset作为一个原子操作实现)。

Kafka如何通过ZooKeeper来进行选举和状态更新？

• 首先Kafka集群启动时，会从Broker中选举一个Controller(分布式锁实现抢先创建临时节点的broker当选)，负责管理集群所有Broker的上下线(监听zk的/brokers/ids/节点)、所有Topic分区副本分配、leader选举等工作。
• 当某个Broker挂了以后，Controller监听到临时节点/brokers/ids/中的变化，从ZK各个分区状态信息中获取ISR(此时去除了挂掉节点所有的Partition，失去leader的Partition重新选举leader)，并完成ZooKeeper各个分区状态更新。

Kafka高效读写的保证？

• Kafka高性能设计包括多分区、顺序读写、page cache、预读、内存映射-零拷贝、无锁offset 管理机制、NIO、压缩、批量读写、高性能序列化（二进制）等。

• 顺序读写：如同HBase顺序写HFile文件一样，Kafka顺序写log文件写入磁盘效率极高(据Kafka官网文档说比随机写快6000倍)。

• 使用操作系统的Page Cache来缓存要写入的数据，好处在于：

  写入前可以做一些优化，提高磁盘写入性能。
• 缓存也可以用于数据被读取，当数据写入与读取速率相近的情况下，可以直接内存读取。
• Page Cache非JVM内存，不会影响JVM，导致GC的增加。同时，Kafka节点宕机，数据还在此机器缓存。

零拷贝机制**：**

正常情况下，先把数据读到内核空间，在从内核空间把数据读到用户空间，然后在调操作系统的io接口写到内核空间，最终在写到硬盘中。磁盘 -> 内核page cache -> 应用缓存 -> 内核page cache -> 磁盘的过程。

• 如果数据只是单纯的拷贝而不需要修改，那么拷贝到应用缓存的步骤完全是多余的。所以Kafka利用了操作系统提供的零拷贝机制，来减少不需要的系统调用和数据拷贝次数，直接在内核空间流转io流，所以kafka的性能非常高。

本篇主要是以Kafka的基础理论、架构原理和面试为主，后续我们再分享Kafka有关 API 以及事务、拦截器、监控等高级篇，已达到Kafka实战编程和应用的目的。