Kafka

Posted 2022-02-06 java_wxid

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Kafka相关的知识，希望对你有一定的参考价值。

—

文章目录

一、Kafka是什么？

定义：Kafka是一个基于zookeeper协调的分布式、多副本的（replica）、支持分区的（partition）系统，它的最大的特性就是可以实时的处理大量数据以满足各种需求场景：比如基于hadoop的批处理系统、低延迟的实时系统、Storm/Spark流式处理引擎，web/nginx日志、访问日志，消息服务等等，用scala语言编写的项目。

二、Kafka的安装与配置

一、Docker安装kafka

Kafka是用Scala语言开发的，运行在JVM上，在安装Kafka之前需要先安装JDK。

yum install java-1.8.0-openjdk* -y

下载zookeeper镜像

docker pull wurstmeister/zookeeper

开放端口

firewall-cmd --add-port=2181/tcp --permanent
firewall-cmd --reload
firewall-cmd --query-port=2181/tcp
systemctl restart docker

启动镜像生成容器

docker run -dit --restart=always --log-driver json-file --log-opt max-size=100m --log-opt max-file=2  --name zookeeper \\
-p 2181:2181 \\
-v /etc/localtime:/etc/localtime \\
-t wurstmeister/zookeeper

下载kafka镜像

docker pull wurstmeister/kafka

开放端口

firewall-cmd --add-port=9092/tcp --permanent
firewall-cmd --reload
firewall-cmd --query-port=9092/tcp
systemctl restart docker

启动kafka镜像生成容器

docker run -dit --restart=always --log-driver json-file --log-opt max-size=100m --log-opt max-file=2 --name kafka \\
-p 9092:9092 \\
-e KAFKA_BROKER_ID=0 \\
-e KAFKA_ZOOKEEPER_CONNECT=172.21.17.47:2181 \\
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://172.21.17.47:9092 \\
-e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 wurstmeister/kafka

参数说明：

-e KAFKA_BROKER_ID=0 在kafka集群中，每个kafka都有一个BROKER_ID来区分自己
-e KAFKA_ZOOKEEPER_CONNECT=ip:2181/kafka 配置zookeeper管理kafka的路径ip:2181/kafka,ip地址改为内网ip
-e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://ip:9092 把kafka的地址端口注册给zookeeper，ip地址改成内网ip
-e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 配置kafka的监听端口
-v /etc/localtime:/etc/localtime 容器时间同步虚拟机的时间

验证kafka是否可以使用

进入容器

docker exec -it kafka bash

进入 /opt/kafka_2.13-2.7.1/bin/ 目录下

cd /opt/kafka_2.13-2.7.1/bin/

运行kafka生产者发送消息

./kafka-console-producer.sh --broker-list 106.14.132.94:9092 --topic sun

发送消息

"datas["channel":"","metric":"temperature","producer":"ijinus","sn":"IJA0101-00002245","time":"1543207156000","value":"80"],"ver":"1.0"

运行kafka消费者接收消息

./kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --topic sun --from-beginning

停止zookeeper和kafka

docker stop zookeeperdocker rm zookeeperdocker stop kafkadocker rm kafka

二、安装包安装kafka

一、安装JDK

yum install java-1.8.0-openjdk* -y

二、安装Zookeeper

wget https://archive.apache.org/dist/zookeeper/zookeeper-3.5.8/apache-zookeeper-3.5.8-bin.tar.gztar -zxvf apache-zookeeper-3.5.8-bin.tar.gzcd apache-zookeeper-3.5.8-bin/cp conf/zoo_sample.cfg conf/zoo.cfgcd /home/apache-zookeeper-3.5.8-bin/bin./zkServer.sh start./zkCli.shls /

打印结果：

[zk: localhost:2181(CONNECTED) 5] ls /[admin, brokers, cluster, config, consumers, controller, controller_epoch, isr_change_notification, kafka, latest_producer_id_block, log_dir_event_notification, zookeeper][zk: localhost:2181(CONNECTED) 6]

三、安装Kafka

wget https://archive.apache.org/dist/kafka/2.4.1/kafka_2.11-2.4.1.tgztar -xzf kafka_2.11-2.4.1.tgzcd /home/kafka_2.11-2.4.1/configvim config/server.properties

配置文件编辑

#broker.id属性在kafka集群中必须要是唯一broker.id=0#kafka部署的机器ip和提供服务的端口号listeners=PLAINTEXT://内网ip:9092   #kafka的消息存储文件log.dir=/usr/local/data/kafka-logs#kafka连接zookeeper的地址zookeeper.connect=内网ip:2181

如果填写外网ip可能会遇上这种情况：

四、启动并验证kafka

启动kafka

/home/kafka_2.11-2.4.1/bin/kafka-server-start.sh config/server.properties &

进入zookeeper目录通过zookeeper客户端查看下zookeeper的目录树

/home/apache-zookeeper-3.5.8-bin/bin/zkCli.shls /ls /brokers/ids

校验kafka

创建主题

/home/kafka_2.11-2.4.1/bin/kafka-topics.sh --create --zookeeper 106.14.132.94:2181 --replication-factor 1 --partitions 1 --topic test

查看kafka中目前存在的topic

/home/kafka_2.11-2.4.1/bin/kafka-topics.sh --list --zookeeper 106.14.132.94:2181

发送消息

/home/kafka_2.11-2.4.1/bin/kafka-console-producer.sh --broker-list 106.14.132.94:9092 --topic test

this is a msg

消费消息

/home/kafka_2.11-2.4.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092--topic test

消费之前的消息

/home/kafka_2.11-2.4.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --from-beginning --topic test

通过jps命令查看运行的情况

对于kafka来说，一个单独的broker意味着kafka集群中只有一个节点。要想增加kafka集群中的节点数量，只需要多启动几个broker实例即可。为了有更好的理解，现在我们在一台机器上同时启动三个broker实例。

三、集群搭建与使用

一、集群配置

首先，我们需要建立好其他2个broker的配置文件：

cp config/server.properties config/server-1.propertiescp config/server.properties config/server-2.properties

配置文件的需要修改的内容分别如下：
config/server-1.properties:

#broker.id属性在kafka集群中必须要是唯一broker.id=1#kafka部署的机器ip和提供服务的端口号listeners=PLAINTEXT://106.14.132.94:9093   log.dir=/usr/local/data/kafka-logs-1#kafka连接zookeeper的地址，要把多个kafka实例组成集群，对应连接的zookeeper必须相同zookeeper.connect=106.14.132.94:2181

config/server-2.properties:

broker.id=2listeners=PLAINTEXT://106.14.132.94:9094log.dir=/usr/local/data/kafka-logs-2zookeeper.connect=106.14.132.94:2181

目前我们已经有一个zookeeper实例和一个broker实例在运行了，现在我们只需要在启动2个broker实例即可：

bin/kafka-server-start.sh -daemon config/server-1.propertiesbin/kafka-server-start.sh -daemon config/server-2.properties

二、验证

查看zookeeper确认集群节点是否都注册成功：

ls /brokers/ids

现在我们创建一个新的topic，副本数设置为3，分区数设置为2：

bin/kafka-topics.sh --create --zookeeper 106.14.132.94:2181 --replication-factor 3 --partitions 2 --topic my-replicated-topic

查看下topic的情况

 bin/kafka-topics.sh --describe --zookeeper 106.14.132.94:2181 --topic my-replicated-topic

以下是输出内容的解释，第一行是所有分区的概要信息，之后的每一行表示每一个partition的信息。
leader节点负责给定partition的所有读写请求，同一个主题不同分区leader副本一般不一样(为了容灾)
replicas 表示某个partition在哪几个broker上存在备份。不管这个几点是不是”leader“，甚至这个节点挂了，也会列出。
isr 是replicas的一个子集，它只列出当前还存活着的，并且已同步备份了该partition的节点。

现在我们向新建的 my-replicated-topic 中发送一些message，kafka集群可以加上所有kafka节点：

bin/kafka-console-producer.sh --broker-list 106.14.132.94:9092,106.14.132.94:9093,106.14.132.94:9094 --topic my-replicated-topic

my test msg 1
my test msg 2

现在开始消费：

bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092,106.14.132.94:9093,106.14.132.94:9094 --from-beginning --topic my-replicated-topic

my test msg 1
my test msg 2

现在我们来测试我们容错性，因为broker1目前是my-replicated-topic的分区0的leader，所以我们要将其kill

ps -ef | grep server.propertieskill 14776

现在再执行命令：

bin/kafka-topics.sh --describe --zookeeper 106.14.132.94:9092 --topic my-replicated-topic

我们可以看到，分区0的leader节点已经变成了broker 0。要注意的是，在Isr中，已经没有了1号节点。leader的选举也是从ISR(in-sync replica)中进行的。
此时，我们依然可以消费新消息：

bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092,106.14.132.94:9093,106.14.132.94:9094 --from-beginning --topic my-replicated-topic

my test msg 1
my test msg 2

查看主题分区对应的leader信息：

get /brokers/topics/my-relicated-topic/partitions/0/state

kafka将很多集群关键信息记录在zookeeper里，保证自己的无状态，从而在水平扩容时非常方便。

三、集群消费

log的partitions分布在kafka集群中不同的broker上，每个broker可以请求备份其他broker上partition上的数据。kafka集群支持配置一个partition备份的数量。针对每个partition，都有一个broker起到“leader”的作用，0个或多个其他的broker作为“follwers”的作用。leader处理所有的针对这个partition的读写请求，而followers被动复制leader的结果，不提供读写(主要是为了保证多副本数据与消费的一致性)。如果这个leader失效了，其中的一个follower将会自动的变成新的leader。

Producers
生产者将消息发送到topic中去，同时负责选择将message发送到topic的哪一个partition中。通过round-robin做简单的负载均衡。也可以根据消息中的某一个关键字来进行区分。通常第二种方式使用的更多。

Consumers
传统的消息传递模式有2种：队列( queue) 和（publish-subscribe）

queue模式：多个consumer从服务器中读取数据，消息只会到达一个consumer。
publish-subscribe模式：消息会被广播给所有的consumer。

Kafka基于这2种模式提供了一种consumer的抽象概念：消费组（consumer group）。

queue模式：所有的consumer都位于同一个consumer group 下。
publish-subscribe模式：所有的consumer都有着自己唯一的consumer group。

上图说明：由2个broker组成的kafka集群，某个主题总共有4个partition(P0-P3)，分别位于不同的broker上。这个集群由2个Consumer Group消费， A有2个consumer instances ，B有4个。通常一个topic会有几个consumer group，每个consumer group都是一个逻辑上的订阅者（ logical subscriber ）。每个consumer group由多个consumer instance组成，从而达到可扩展和容灾的功能。

四、Kafka可视化管理工具kafka-manager

安装及基本使用可参考：Java廖志伟

三、Kafka的使用场景

日志收集：一个公司可以用Kafka收集各种服务的log，通过kafka以统一接口服务的方式开放给各种consumer，例如hadoop、Hbase、Solr等。
消息系统：解耦和生产者和消费者、缓存消息等。
用户活动跟踪：Kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后订阅者通过订阅这些topic来做实时的监控分析，或者装载到hadoop、数据仓库中做离线分析和挖掘。
运营指标：Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。

四、Kafka基本概念

Broker：消息中间件处理节点，一个Kafka节点就是一个broker，一个或者多个Broker可以组成一个Kafka集群。
Topic：Kafka根据topic对消息进行归类，发布到Kafka集群的每条消息都需要指定一个topic。
Producer：消息生产者，向Broker发送消息的客户端。 Consumer：消息消费者，从Broker读取消息的客户端。
ConsumerGroup：每个Consumer属于一个特定的Consumer Group，一条消息可以被多个不同的ConsumerGroup消费，但是一个Consumer Group中只能有一个Consumer能够消费该消息。
Partition：物理上的概念，一个topic可以分为多个partition，每个partition内部消息是有序的。

五、Kafka的初体验

创建主题

创建一个名字为“test”的Topic，这个topic只有一个partition，并且备份因子也设置为1。当producer发布一个消息到某个指定的Topic，这个Topic如果不存在，就自动创建。

/opt/kafka_2.13-2.7.1/bin/kafka-topics.sh --create --zookeeper 106.14.132.94:2181 --replication-factor 1 --partitions 1 --topic test

查看kafka中目前存在的topic

/opt/kafka_2.13-2.7.1/bin/kafka-topics.sh --list --zookeeper 106.14.132.94:2181

列表中有一个__consumer_offsets主题，这个主题不能删除哟

删除主题

/opt/kafka_2.13-2.7.1/bin/kafka-topics.sh --delete --topic test --zookeeper 106.14.132.94:2181

如果出现This will have no impact if delete.topic.enable is not set to true.

彻底删除topic：
[1. ] 删除Topic，delete.topic.enable=true这里要设置为true，需要在$KAFKA/config/server.properties中配置delete.topic.enble=true
[2. ] 删除log日志
[3. ] 删除ZK中的Topic记录

删除列表中有一个**consumer_offsets主题会出现，Topic **consumer_offsets is a kafka internal topic and is not allowed to be marked for deletion.

__consumer_offsets这个topic是由kafka自动创建的，默认49个，这个topic是不能被删除的！

为什么这里会是这样存储__consumer_offsets的呢？

[1.] 将所有 N Broker 和待分配的 i 个 Partition 排序[2.] 将第 i 个 Partition 分配到第(i mod n)个 Broker 上

发送消息

kafka自带了一个producer命令客户端，可以从本地文件中读取内容，或者我们也可以以命令行中直接输入内容，并将这些内容以消息的形式发送到kafka集群中。在默认情况下，每一个行会被当做成一个独立的消息。首先我们要运行发布消息的脚本，然后在命令中输入要发送的消息的内容。

/opt/kafka_2.13-2.7.1/bin/kafka-console-producer.sh --broker-list 106.14.132.94:9092 --topic test

this is a news
this is a another news

消费消息

对于consumer，kafka同样也携带了一个命令行客户端，会将获取到内容在命令中进行输出，默认是消费最新的消息。由于我们已经发送了消息了，想要消费之前的消息可以通过–from-beginning参数指定。

消费之前的消息

/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --from-beginning --topic test

消费最新的消息

/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --topic test

通过不同的终端窗口来运行以上的命令，你将会看到在producer终端输入的内容，很快就会在consumer的终端窗口上显示出来。

消费之前多主题的消息

/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --from-beginning --whitelist "test|test2"

六、消费模式

单播消费

一条消息只能被某一个消费者消费的模式，类似queue模式，只需让所有消费者在同一个消费组里即可。

分别在两个客户端执行如下消费命令，然后往主题里发送消息，结果只有一个客户端能收到消息。
发送消息：/opt/kafka_2.13-2.7.1/bin/kafka-console-producer.sh --broker-list 106.14.132.94:9092 --topic test

客户端1：/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --consumer-property group.id=testGroup --topic test

客户端2：/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --consumer-property group.id=testGroup --topic test

多播消费

一条消息能被多个消费者消费的模式，类似publish-subscribe模式费，针对Kafka同一条消息只能被同一个消费组下的某一个消费者消费的特性，要实现多播只要保证这些消费者属于不同的消费组即可。我们再增加一个消费者，该消费者属于testGroup-2消费组，结果两个客户端都能收到消息。

发送消息：/opt/kafka_2.13-2.7.1/bin/kafka-console-producer.sh --broker-list 106.14.132.94:9092 --topic test

this is a more news
this is a more new2
this is a more new3

客户端1：/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --consumer-property group.id=testGroup --topic test

客户端2：/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --consumer-property group.id=testGroup --topic test

客户端3：/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092 --consumer-property group.id=testGroup2 --topic test

七、消费组的偏移量（offset）

查看消费组名

/opt/kafka_2.13-2.7.1/bin/kafka-consumer-groups.sh --bootstrap-server 106.14.132.94:9092 --list

查看消费组的消费偏移量

查看消费组testGroup和testGroup2的消费偏移量

/opt/kafka_2.13-2.7.1/bin/kafka-consumer-groups.sh --bootstrap-server 106.14.132.94:9092 --describe --group testGroup

/opt/kafka_2.13-2.7.1/bin/kafka-consumer-groups.sh --bootstrap-server 106.14.132.94:9092 --describe --group testGroup2

消费组的消费偏移量参数

current-offset：当前消费组的已消费偏移量
log-end-offset：主题对应分区消息的结束偏移量(HW)
lag：当前消费组未消费的消息数

发送消息理解消费组的消费偏移量

1个客户端的消费组是testGroup，一个客户端的消费组是testGroup，他们二个都是同一个test主题。

发送一条消息

/opt/kafka_2.13-2.7.1/bin/kafka-console-producer.sh --broker-list 106.14.132.94:9092 --topic test

this is a testgroup message

查看偏移量

/opt/kafka_2.13-2.7.1/bin/kafka-consumer-groups.sh --bootstrap-server 106.14.132.94:9092 --describe --group testGroup

当前消费组的已消费偏移量（current-offset）没变，说明当前的消费组没有消费者进行消费，因为没有启动消费者。主题对应分区消息的结束偏移量（log-end-offset）加1了，说明当前分区的消息多了一条。当前消费组未消费的消息数（lag）加1了，说明有1条消息没有被消费。

客户端的消费组testGroup开始消费

/opt/kafka_2.13-2.7.1/bin/kafka-console-consumer.sh --bootstrap-server 106.14.132.94:9092  --consumer-property group.id=testGroup --topic test

再次查看消费组testGroup和testGroup2的消费偏移量

/opt/kafka_2.13-2.7.1/bin/kafka-consumer-groups.sh --bootstrap-server 106.14.132.94:9092 --describe --group testGroup

/opt/kafka_2.13-2.7.1/bin/kafka-consumer-groups.sh --bootstrap-server 106.14.132.94:9092 --describe --group testGroup2

发现testGroup消费组当前消费组的已消费偏移量（current-offset）加1了，当前消费组未消费的消息数（lag）减1了，说明testGroup已经被消费了。testGroup2消费者还是一样没有发生变化，所以可以推测出，消费者是通过不同的消费组进行消费的，每个消费组互不影响。

八、主题/分区/日志的概念

主题

Topic是一个类别的名称，同类消息发送到同一个Topic下面。对于每一个Topic，下面可以有多个分区(Partition)日志文件。Topic是一个逻辑概念，真正落实的数据都在分区上面，一般每个主题至少有一个分区。这样可以通过多个分区，放在在不同的服务器上面去，达到分布式存储的功能，海量数据通过分区存储，切片存储解决单台机器无法存储海量数据的问题。

分区

分区是一个有序的消息序列，这些消息按顺序添加到一个叫做commit log的文件中。每个分区中的消息都有一个唯一的编号，称之为offset，用来唯一标示某个分区中的消息。每个分区，都对应一个commit log文件。一个分区中的消息的offset都是唯一的，但是不同的分区中的消息的offset可能是相同的。

日志

kafka一般不会删除消息，不管这些消息有没有被消费。只会根据配置的日志保留时间(log.retention.hours)确认消息多久被删除，默认保留最近一周的日志消息。kafka的性能与保留的消息数据量大小没有关系，因此保存大量的数据消息日志信息不会有什么影响。

每个消费者是基于自己在commit log中的消费进度(offset)来进行工作的。在kafka中，消费offset由消费者自己来维护；一般情况下我们按照顺序逐条消费commit log中的消息，当然我可以通过指定offset来重复消费某些消息，或者跳过某些消息。这意味kafka中的消费者对集群的影响是非常小的，添加一个或者减少一个消费者，对于集群或者其他消费者来说，都是没有影响的，因为每个消费者维护各自的消费offset。

九、Topic/Partition/Broker的初体验

创建多个分区的主题

/opt/kafka_2.13-2.7.1/bin/kafka-topics.sh --create --zookeeper 106.14.132.94:2181 --replication-factor 1 --partitions 2 --topic test1

查看下topic的情况

/opt/kafka_2.13-2.7.1/bin/kafka-topics.sh --describe --zookeeper 106.14.132.94:2181 --topic test1

第一行是所有分区的概要信息，之后的每一行表示每一个partition的信息。

leader节点负责给定partition的所有读写请求。
replicas表示某个partition在哪几个broker上存在备份。不管这个几点是不是”leader“，甚至这个节点挂了，也会列出。
isr是replicas的一个子集，它只列出当前还存活着的，并且已同步备份了该partition的节点。

创建的名称为"test"的topic

/opt/kafka_2.13-2.7.1/bin/kafka-topics.sh --describe --zookeeper 106.14.132.94:2181 --topic test

之前设置了topic的partition数量为1，备份因子为1，因此显示就如上所示了。

查询kafka的数据文件存储目录

cat /opt/kafka_2.13-2.7.1/config/server.properties

找到log.dirs配置的路径

进入kafka的数据文件存储目录查看test和test1主题的消息日志文件

消息日志文件主要存放在分区文件夹里的以log结尾的日志文件里

增加topic的分区数量(目前kafka不支持减少分区)

/opt/kafka_2.13-2.7.1/bin/kafka-topics.sh -alter --partitions 3 --zookeeper 106.14.132.94:2181 --topic test

查看分区

/opt/kafka_2.13-2.7.1/bin/kafka-topics.sh --describe --zookeeper 106.14.132.94:2181 --topic test

一个主题，代表逻辑上的一个业务数据集，比如按数据库里不同表的数据操作消息区分放入不同主题，订单相关操作消息放入订单主题，用户相关操作消息放入用户主题，对于大型网站来说，后端数据都是海量的，订单消息很可能是非常巨量的，比如有几百个G甚至达到TB级别，如果把这么多数据都放在一台机器上可定会有容量限制问题，那么就可以在主题内部划分多个分区来分片存储数据，不同的分区可以位于不同的机器上，每台机器上都运行一个Kafka的进程Broker。

为什么要对主题下数据进行分区存储？

commit log文件会受到所在机器的文件系统大小的限制，分区之后可以将不同的分区放在不同的机器上，相当于对数据做了分布式存储，理论上一个主题可以处理任意数量的数据。
为了提高并行度。

十、消费顺序

一个分区同一个时刻在一个消费组中只能有一个消费者实例在消费，从而保证消费顺序。消费组中的消费者实例的数量不能比一个主题中的分区的数量多，否则，多出来的消费者消费不到消息。Kafka只在分区的范围内保证消息消费的局部顺序性，不能在同一个主题中的多个分区中保证总的消费顺序性。如果有在总体上保证消费顺序的需求，那么我们可以通过将主题的分区数量设置为1，将消费组中的消费者实例数量也设置为1，但是这样会影响性能，所以kafka的顺序消费很少用。

十一、Kafka核心总控制器Controller

在Kafka集群中会有一个或者多个broker，其中有一个broker会被选举为控制器（Kafka Controller），它负责管理整个集群中所有分区和副本的状态。

当某个分区的leader副本出现故障时，由控制器负责为该分区选举新的leader副本。
当检测到某个分区的ISR集合发生变化时，由控制器负责通知所有broker更新其元数据信息。
当使用kafka-topics.sh脚本为某个topic增加分区数量时，同样还是由控制器负责让新分区被其他节点感知到。

十二、Controller选举机制

在kafka集群启动的时候，会自动选举一台broker作为总控制器来管理整个集群，选举的过程是集群中每个broker都会尝试在zookeeper上创建一个/controller临时节点，zookeeper会保证只有一个broker能创建成功，这个broker就会成为集群的总控制器。当这个总控制器角色的broker宕机了，此时zookeeper临时节点会消失，集群里其他broker会一直监听这个临时节点，发现临时节点消失了，就竞争再次创建临时节点，就是我们上面说的选举机制，zookeeper又会保证有一个broker成为新的controller。

具备控制器身份的broker需要比其他普通的broker多一份职责，具体细节如下：

监听broker相关的变化。为Zookeeper中的/brokers/ids/节点添加BrokerChangeListener，用来处理broker增减的变化。
监听topic相关的变化。为Zookeeper中的/brokers/topics节点添加TopicChangeListener，用来处理topic增减的变化；为Zookeeper中的/admin/delete_topics节点添加TopicDeletionListener，用来处理删除topic的动作。
从Zookeeper中读取获取当前所有与topic、partition以及broker有关的信息并进行相应的管理。对于所有topic所对应的Zookeeper中的/brokers/topics/[topic]节点添加PartitionModificationsListener，用来监听topic中的分区分配变化。
更新集群的元数据信息，同步到其他普通的broker节点中。

十三、Partition副本选举Leader机制

控制器感知到分区leader所在的broker挂了(控制器监听了很多zk节点可以感知到broker存活)，控制器会从ISR列表(参数unclean.leader.election.enable=false的前提下)里挑第一个broker作为leader(第一个broker最先放进ISR列表，可能是同步数据最多的副本)，如果参数unclean.leader.election.enable为true，代表在ISR列表里所有副本都挂了的时候可以在ISR列表以外的副本中选leader，这种设置，可以提高可用性，但是选出的新leader有可能数据少很多。

副本进入ISR列表有两个条件：

副本节点不能产生分区，必须能与zookeeper保持会话以及跟leader副本网络连通
副本能复制leader上的所有写操作，并且不能落后太多。(与leader副本同步滞后的副本，是由replica.lag.time.max.ms 配置决定的，超过这个时间都没有跟leader同步过的一次的副本会被移出ISR列表)

十四、消费者消费消息的offset记录机制

每个消费者会定期将自己消费分区的offset提交给kafka内部topic：consumer_offsets，提交过去的时候，key是consumerGroupId+topic+分区号，value就是当前offset的值，kafka会定期清理topic里的消息，最后就保留最新的那条数据因为consumer_offsets可能会接收高并发的请求，kafka默认给其分配50个分区(可以通过offsets.topic.num.partitions设置)，这样可以通过加机器的方式抗大并发。通过如下公式可以选出consumer消费的offset要提交到**consumer_offsets的哪个分区，公式：hash(consumerGroupId) % **consumer_offsets主题的分区数

十五、消费者Rebalance机制

rebalance就是说如果消费组里的消费者数量有变化或消费的分区数有变化，kafka会重新分配消费者消费分区的关系。比如consumer
group中某个消费者挂了，此时会自动把分配给他的分区交给其他的消费者，如果他又重启了，那么又会把一些分区重新交还给他。

注意：rebalance只针对subscribe这种不指定分区消费的情况，如果通过assign这种消费方式指定了分区，kafka不会进行rebanlance。
如下情况可能会触发消费者rebalance：

消费组里的consumer增加或减少了。
动态给topic增加了分区
消费组订阅了更多的topic。

rebalance过程中，消费者无法从kafka消费消息，这对kafka的TPS会有影响，如果kafka集群内节点较多，比如数百个，那重平衡可能会耗时极多，所以应尽量避免在系统高峰期的重平衡发生。

十六、消费者Rebalance分区分配策略

主要有三种rebalance的策略：range、round-robin、sticky。
Kafka 提供了消费者客户端参数partition.assignment.strategy 来设置消费者与订阅主题之间的分区分配策略。默认情况为range分配策略。
假设一个主题有10个分区(0-9)，现在有三个consumer消费：
range策略就是按照分区序号排序，假设 n＝分区数／消费者数量 = 3， m＝分区数%消费者数量 = 1，那么前 m 个消费者每个分配 n+1 个分区，后面的（消费者数量－m ）个消费者每个分配 n 个分区。
比如分区03给一个consumer，分区46给一个consumer，分区7~9给一个consumer。
round-robin策略就是轮询分配，比如分区0、3、6、9给一个consumer，分区1、4、7给一个consumer，分区2、5、8给一个consumer
sticky策略初始时分配策略与round-robin类似，但是在rebalance的时候，需要保证如下两个原则。
1）分区的分配要尽可能均匀。
2）分区的分配尽可能与上次分配的保持相同。
当两者发生冲突时，第一个目标优先于第二个目标。这样可以最大程度维持原来的分区分配的策略。
比如对于第一种range情况的分配，如果第三个consumer挂了，那么重新用sticky策略分配的结果如下：
consumer1除了原有的0~3，会再分配一个7
consumer2除了原有的4~6，会再分配8和9

十七、Rebalance过程

第一阶段：选择组协调器

组协调器GroupCoordinator

每个消费组都会选择一个broker作为自己的组协调器coordinator，负责监控这个消费组里的所有消费者的心跳，以及判断是否宕机，然后开启消费者rebalance。消费组中的每个消费者启动时会向kafka集群中的某个节点发送 FindCoordinatorRequest 请求来查找对应的组协调器GroupCoordinator，并跟其建立网络连接。

组协调器选择方式

消费者消费的offset要提交到__consumer_offsets的哪个分区，这个分区leader对应的broker就是这个消费组的coordinator

第二阶段：加入消费组JOIN GROUP

在成功找到消费组所对应的 GroupCoordinator 之后就进入加入消费组的阶段，在此阶段的消费者会向 GroupCoordinator 发送 JoinGroupRequest 请求，并处理响应。然后GroupCoordinator 从一个consumer group中选择第一个加入group的consumer作为leader(消费组协调器)，把consumer group情况发送给这个leader，接着这个leader会负责制定分区方案。

第三阶段（ SYNC GROUP)

consumer leader通过给GroupCoordinator发送SyncGroupRequest，接着GroupCoordinator就把分区方案下发给各个消费者，他们会根据指定分区的leader broker进行网络连接以及消息消费。

十八、producer发布消息机制

写入方式

producer 采用 push 模式将消息发布到 broker，每条消息都被 append 到 patition 中，属于顺序写磁盘（顺序写磁盘效率比随机写内存要高，保障 kafka 吞吐率）。

消息路由

producer 发送消息到 broker 时，会根据分区算法选择将其存储到哪一个 partition。其路由机制为：

指定了 patition，则直接使用；
未指定 patition 但指定 key，通过对 key 的 value 进行hash 选出一个 patition
patition 和 key 都未指定，使用轮询选出一个 patition。

写入流程

producer 先从 zookeeper 的 “/brokers/…/state” 节点找到该 partition 的 leader。
producer 将消息发送给该 leader。
leader 将消息写入本地 log。
followers 从 leader pull 消息，写入本地 log 后向leader 发送 ACK。
leader 收到所有 ISR 中的 replica 的 ACK 后，增加 HW（high watermark，最后 commit 的 offset）并向 producer 发送 ACK。

十九、HW与LEO

HW俗称高水位，HighWatermark的缩写，取一个partition对应的ISR中最小的LEO(log-end-offset)作为HW，consumer最多只能消费到HW所在的位置。另外每个replica都有HW,leader和follower各自负责更新自己的HW的状态。对于leader新写入的消息，consumer不能立刻消费，leader会等待该消息被所有ISR中的replicas同步后更新HW，此时消息才能被consumer消费。这样就保证了如果leader所在的broker失效，该消息仍然可以从新选举的leader中获取。对于来自内部broker的读取请求，没有HW的限制。

下图详细的说明了当producer生产消息至broker后，ISR以及HW和LEO的流转过程

由此可见，Kafka的复制机制既不是完全的同步复制，也不是单纯的异步复制。事实上，同步复制要求所有能工作的follower都复制完，这条消息才会被commit，这种复制方式极大的影响了吞吐率。而异步复制方式下，follower异步的从leader复制数据，数据只要被leader写入log就被认为已经commit，这种情况下如果follower都还没有复制完，落后于leader时，突然leader宕机，则会丢失数据。而Kafka的这种使用ISR的方式则很好的均衡了确保数据不丢失以及吞吐率。再回顾下消息发送端对发出消息持久化机制参数acks的设置，我们结合HW和LEO来看下acks=1的情况。

结合HW和LEO看下 acks=1的情况

二十、日志分段存储

Kafka 一个分区的消息数据对应存储在一个文件夹下，以topic名称+分区号命名，消息在分区内是分段(segment)存储，每个段的消息都存储在不一样的log文件里，这种特性方便oldsegment file快速被删除，kafka规定了一个段位的 log 文件最大为 1G，做这个限制目的是为了方便把 log文件加载到内存去操作。

# 部分消息的offset索引文件，kafka每次往分区发4K(可配置)消息就会记录一条当前消息的offset到index文件，# 如果要定位消息的offset会先在这个文件里快速定位，再去log文件里找具体消息00000000000000000000.index# 消息存储文件，主要存offset和消息体00000000000000000000.log# 消息的发送时间索引文件，kafka每次往分区发4K(可配置)消息就会记录一条当前消息的发送时间戳与对应的offset到timeindex文件，# 如果需要按照时间来定位消息的offset，会先在这个文件里查找00000000000000000000.timeindex00000000000005367851.index00000000000005367851.log00000000000005367851.timeindex00000000000009936472.index00000000000009936472.log00000000000009936472.timeindex

这个 9936472 之类的数字，就是代表了这个日志段文件里包含的起始 Offset，也就说明这个分区里至少都写入了接近 1000万条数据了。Kafka Broker 有一个参数，log.segment.bytes，限定了每个日志段文件的大小，最大就是 1GB。一个日志段文件满了，就自动开一个新的日志段文件来写入，避免单个文件过大，影响文件的读写性能，这个过程叫做 log rolling，正在被写入的那个日志段文件，叫做 active log segment。

二十一、十亿消息数据线上环境规划

二十二、JVM参数设置

kafka是scala语言开

以上是关于Kafka的主要内容，如果未能解决你的问题，请参考以下文章

Kafka（四）集群之kafka

kafka NoClassDefFoundError kafka / Kafka

解开Kafka神秘的面纱：kafka优雅应用

kafka NoClassDefFoundError kafka/Kafka

Kafka原理及Kafka群集部署